From 1f3018f98d7fa42b18673171da818b719b9decf7 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 3 Jul 2026 17:56:29 -0500 Subject: [PATCH 1/9] widen-system-uid task groups 1,3,4,5: Akka.Remote int->long system uid - AddressUid.Uid + AddressUidExtension.Uid(ActorSystem) widened to long; default generation unchanged (ThreadLocalRandom int-range, rolling-upgrade safe) - new akka.remote.use-64bit-system-uids switch (default off) enables full-range nonzero 64-bit uid generation via netstandard2.0-safe byte-based RNG - HandshakeInfo/refuseUid, EndpointRegistry, Endpoint.cs (HopelessAssociation, ReliableDeliverySupervisor, GotUid, EndpointWriter/Reader), EndpointManager messages (Pass/Quarantined/Quarantine/ResendState) re-typed long/long? - RemoteWatcher HeartbeatRsp/_addressUids/Quarantine re-typed long - quarantine API hard re-type: IRemoteActorRefProvider/RemoteActorRefProvider/ RemoteTransport/Remoting.Quarantine(Address, long?); QuarantinedEvent(Address, long) - MiscMessageSerializer + AkkaPduCodec 32-bit narrowing casts removed (wire already 64-bit: ContainerFormats uint64, WireFormats fixed64) Akka.Cluster intentionally left broken until the next task group commit. --- .../widen-system-uid-to-64bit/tasks.md | 26 ++++---- .../AddressUidExtensionSpecs.cs | 2 +- .../Akka.Remote.Tests/RemoteWatcherSpec.cs | 8 +-- src/core/Akka.Remote/AddressUidExtension.cs | 63 +++++++++++++++---- .../Akka.Remote/Configuration/Remote.conf | 6 ++ src/core/Akka.Remote/Endpoint.cs | 30 ++++----- src/core/Akka.Remote/EndpointManager.cs | 18 +++--- src/core/Akka.Remote/EndpointRegistry.cs | 20 +++--- .../Akka.Remote/RemoteActorRefProvider.cs | 4 +- src/core/Akka.Remote/RemoteTransport.cs | 2 +- src/core/Akka.Remote/RemoteWatcher.cs | 16 ++--- src/core/Akka.Remote/Remoting.cs | 2 +- .../Akka.Remote/RemotingLifecycleEvent.cs | 4 +- .../Serialization/MiscMessageSerializer.cs | 4 +- .../Akka.Remote/Transport/AkkaPduCodec.cs | 2 +- .../Transport/AkkaProtocolTransport.cs | 20 +++--- 16 files changed, 137 insertions(+), 90 deletions(-) diff --git a/openspec/changes/widen-system-uid-to-64bit/tasks.md b/openspec/changes/widen-system-uid-to-64bit/tasks.md index b30759afeae..2f4bddfaf21 100644 --- a/openspec/changes/widen-system-uid-to-64bit/tasks.md +++ b/openspec/changes/widen-system-uid-to-64bit/tasks.md @@ -1,8 +1,8 @@ ## 1. Core type + generation -- [ ] 1.1 Widen `AddressUid.Uid` (field) + `AddressUidExtension.Uid(ActorSystem)` `int → long` -- [ ] 1.2 Add a netstandard2.0-safe 64-bit RNG; default generation stays in `[0, int.MaxValue]` -- [ ] 1.3 Config switch to enable full 64-bit uid generation (default off; documented "all nodes v1.6 first" precondition) +- [x] 1.1 Widen `AddressUid.Uid` (field) + `AddressUidExtension.Uid(ActorSystem)` `int → long` +- [x] 1.2 Add a netstandard2.0-safe 64-bit RNG; default generation stays in `[0, int.MaxValue]` +- [x] 1.3 Config switch to enable full 64-bit uid generation (default off; documented "all nodes v1.6 first" precondition) — `akka.remote.use-64bit-system-uids` ## 2. Cluster @@ -12,27 +12,27 @@ ## 3. Remoting state machine (internal `int → long`) -- [ ] 3.1 `HandshakeInfo` ctor + `Uid`; `AkkaProtocolTransport` refuseUid path -- [ ] 3.2 `EndpointRegistry` (register/quarantine/refuseUid) -- [ ] 3.3 `Endpoint.cs`: `HopelessAssociation`, `ReliableDeliverySupervisor`, `GotUid`, `EndpointWriter` uid fields/params -- [ ] 3.4 `EndpointManager` messages: `Pass`, `Quarantined`, `Quarantine`, `ResendState` +- [x] 3.1 `HandshakeInfo` ctor + `Uid`; `AkkaProtocolTransport` refuseUid path +- [x] 3.2 `EndpointRegistry` (register/quarantine/refuseUid) +- [x] 3.3 `Endpoint.cs`: `HopelessAssociation`, `ReliableDeliverySupervisor`, `GotUid`, `EndpointWriter` uid fields/params +- [x] 3.4 `EndpointManager` messages: `Pass`, `Quarantined`, `Quarantine`, `ResendState` ## 4. RemoteWatcher -- [ ] 4.1 `HeartbeatRsp(long addressUid)` + `long AddressUid`; `_addressUids: Dictionary`; `ReceiveHeartbeatRsp` -- [ ] 4.2 `Quarantine(Address, long? addressUid)` +- [x] 4.1 `HeartbeatRsp(long addressUid)` + `long AddressUid`; `_addressUids: Dictionary`; `ReceiveHeartbeatRsp` +- [x] 4.2 `Quarantine(Address, long? addressUid)` ## 5. Quarantine API (hard re-type) -- [ ] 5.1 `IRemoteActorRefProvider.Quarantine(Address, long? uid)` -- [ ] 5.2 `RemoteActorRefProvider.Quarantine`, `RemoteTransport.Quarantine`, `Remoting.Quarantine` -- [ ] 5.3 `QuarantinedEvent(Address, long uid)` + `long Uid` +- [x] 5.1 `IRemoteActorRefProvider.Quarantine(Address, long? uid)` +- [x] 5.2 `RemoteActorRefProvider.Quarantine`, `RemoteTransport.Quarantine`, `Remoting.Quarantine` +- [x] 5.3 `QuarantinedEvent(Address, long uid)` + `long Uid` ## 6. Wire + serializers - [ ] 6.1 `ClusterMessages.proto` `UniqueAddress.uid` `uint32 → uint64`; regenerate - [ ] 6.2 Remove narrowing casts: `ClusterMessageSerializer`, `MiscMessageSerializer`, `AkkaPduCodec`, DData `SerializationSupport.UniqueAddressFromProto` -- [ ] 6.3 Coordinate `serializer-v2` schema to emit/read 64-bit uid +- [x] 6.3 Coordinate `serializer-v2` schema to emit/read 64-bit uid (Decision 11 in serializer-v2 design.md) ## 7. API approval + build diff --git a/src/core/Akka.Remote.Tests/AddressUidExtensionSpecs.cs b/src/core/Akka.Remote.Tests/AddressUidExtensionSpecs.cs index 7fabdf2cb81..abe135eda19 100644 --- a/src/core/Akka.Remote.Tests/AddressUidExtensionSpecs.cs +++ b/src/core/Akka.Remote.Tests/AddressUidExtensionSpecs.cs @@ -23,7 +23,7 @@ public class AddressUidExtensionSpecs : AkkaSpec [Fact] public void AddressUidExtension_should_always_report_same_value() { - var values = new ConcurrentBag(); + var values = new ConcurrentBag(); var parallelOps = 1000; var loop = Parallel.For(0, parallelOps, _ => { diff --git a/src/core/Akka.Remote.Tests/RemoteWatcherSpec.cs b/src/core/Akka.Remote.Tests/RemoteWatcherSpec.cs index cb3dc7b780c..2b8b2dcb41d 100644 --- a/src/core/Akka.Remote.Tests/RemoteWatcherSpec.cs +++ b/src/core/Akka.Remote.Tests/RemoteWatcherSpec.cs @@ -77,7 +77,7 @@ public override int GetHashCode() public class Quarantined { - public Quarantined(Address address, int? uid) + public Quarantined(Address address, long? uid) { Address = address; Uid = uid; @@ -85,7 +85,7 @@ public Quarantined(Address address, int? uid) public Address Address { get; } - public int? Uid { get; } + public long? Uid { get; } protected bool Equals(Quarantined other) { @@ -139,7 +139,7 @@ protected override void PublishAddressTerminated(Address address) Context.System.EventStream.Publish(new AddressTerm(address)); } - protected override void Quarantine(Address address, int? addressUid) + protected override void Quarantine(Address address, long? addressUid) { // don't quarantine in remoting, but publish a testable message Context.System.EventStream.Publish(new Quarantined(address, addressUid)); @@ -182,7 +182,7 @@ protected override void AfterAll() private readonly Address _remoteAddress; private readonly RemoteWatcher.HeartbeatRsp _heartbeatRspB; - private int RemoteAddressUid + private long RemoteAddressUid { get { return AddressUidExtension.Uid(_remoteSystem); } } diff --git a/src/core/Akka.Remote/AddressUidExtension.cs b/src/core/Akka.Remote/AddressUidExtension.cs index 1c6be81373a..a851a8c3c39 100644 --- a/src/core/Akka.Remote/AddressUidExtension.cs +++ b/src/core/Akka.Remote/AddressUidExtension.cs @@ -5,6 +5,7 @@ // //----------------------------------------------------------------------- +using System; using Akka.Actor; using Akka.Util; @@ -16,23 +17,25 @@ namespace Akka.Remote public class AddressUidExtension : ExtensionIdProvider { /// - /// TBD + /// Creates the extension for , honoring the + /// akka.remote.use-64bit-system-uids config switch. /// - /// TBD - /// TBD + /// The actor system that owns this extension. + /// A new instance. public override AddressUid CreateExtension(ExtendedActorSystem system) { - return new AddressUid(); + var use64BitUid = system.Settings.Config.GetBoolean("akka.remote.use-64bit-system-uids", false); + return new AddressUid(use64BitUid); } #region Static methods /// - /// TBD + /// Returns the unique identifier for this incarnation of . /// - /// TBD - /// TBD - public static int Uid(ActorSystem system) + /// The actor system whose UID is being retrieved. + /// The address/system UID for this actor system incarnation. + public static long Uid(ActorSystem system) { return system.WithExtension().Uid; } @@ -41,17 +44,55 @@ public static int Uid(ActorSystem system) } /// - /// Extension that holds a UID that is assigned as a random 'Int'. - /// + /// Extension that holds a UID that is assigned as a random . + /// /// The UID is intended to be used together with an to be /// able to distinguish restarted actor system using the same host and port. /// public class AddressUid : IExtension { + /// + /// Creates a new using the legacy int-range (rolling-upgrade safe) generation. + /// + public AddressUid() : this(false) + { + } + + /// + /// Creates a new , optionally generating a full-range 64-bit UID. + /// + /// + /// When true, generates a uniformly random nonzero across the full 64-bit range + /// (negative values allowed). When false (default), generates a value in [0, int.MaxValue] + /// for rolling-upgrade compatibility with pre-v1.6 nodes. + /// + internal AddressUid(bool use64BitUid) + { + Uid = use64BitUid ? Generate64BitUid() : ThreadLocalRandom.Current.Next(); + } + /// /// The random unique identifier for this incarnation of the ActorSystem. /// - public readonly int Uid = ThreadLocalRandom.Current.Next(); + public readonly long Uid; + + /// + /// Generates a uniformly random nonzero across the full 64-bit range. + /// Zero is reserved as a sentinel value and is never returned. netstandard2.0-safe + /// (does not rely on Random.NextInt64(), which is only available on net6.0+). + /// + private static long Generate64BitUid() + { + var buf = new byte[8]; + long candidate; + do + { + ThreadLocalRandom.Current.NextBytes(buf); + candidate = BitConverter.ToInt64(buf, 0); + } while (candidate == 0); + + return candidate; + } } } diff --git a/src/core/Akka.Remote/Configuration/Remote.conf b/src/core/Akka.Remote/Configuration/Remote.conf index 0cc7b1e2fed..68979925340 100644 --- a/src/core/Akka.Remote/Configuration/Remote.conf +++ b/src/core/Akka.Remote/Configuration/Remote.conf @@ -142,6 +142,12 @@ akka { # every module will respect this setting. use-dispatcher = "akka.remote.default-remote-dispatcher" + # When enabled, system/address UIDs are generated across the full 64-bit range + # (nonzero, may be negative) instead of the legacy [0, int.MaxValue] range. + # DO NOT enable until EVERY node in the cluster and all remote peers run Akka.NET v1.6+: + # pre-v1.6 nodes silently truncate UIDs above uint32.MaxValue in cluster gossip. + use-64bit-system-uids = off + ### Security settings # Enable untrusted mode for full security of server managed actors, prevents diff --git a/src/core/Akka.Remote/Endpoint.cs b/src/core/Akka.Remote/Endpoint.cs index 5decec454e3..22184820fff 100644 --- a/src/core/Akka.Remote/Endpoint.cs +++ b/src/core/Akka.Remote/Endpoint.cs @@ -291,7 +291,7 @@ internal sealed class HopelessAssociation : EndpointException, IAssociationProbl /// TBD /// TBD /// TBD - public HopelessAssociation(Address localAddress, Address remoteAddress, int? uid = null, Exception cause = null) + public HopelessAssociation(Address localAddress, Address remoteAddress, long? uid = null, Exception cause = null) : base("Catastrophic association error.", cause) { RemoteAddress = remoteAddress; @@ -312,7 +312,7 @@ public HopelessAssociation(Address localAddress, Address remoteAddress, int? uid /// /// TBD /// - public int? Uid { get; private set; } + public long? Uid { get; private set; } } /// @@ -439,7 +439,7 @@ private TooLongIdle() { } private readonly Address _localAddress; private readonly Address _remoteAddress; - private readonly int? _refuseUid; + private readonly long? _refuseUid; private readonly AkkaProtocolTransport _transport; private readonly RemoteSettings _settings; private readonly AkkaPduCodec _codec; @@ -461,7 +461,7 @@ public ReliableDeliverySupervisor( AkkaProtocolHandle handleOrActive, Address localAddress, Address remoteAddress, - int? refuseUid, + long? refuseUid, AkkaProtocolTransport transport, RemoteSettings settings, AkkaPduCodec codec, @@ -495,7 +495,7 @@ public ReliableDeliverySupervisor( /// /// The UID of the remote system we're communicating with. /// - public int? Uid { get; set; } + public long? Uid { get; set; } /// /// Processing of s has to be delayed until the UID is discovered after a reconnect. Depending whether the @@ -810,7 +810,7 @@ public sealed class GotUid /// /// TBD /// TBD - public GotUid(int uid, Address remoteAddress) + public GotUid(long uid, Address remoteAddress) { Uid = uid; RemoteAddress = remoteAddress; @@ -819,7 +819,7 @@ public GotUid(int uid, Address remoteAddress) /// /// TBD /// - public int Uid { get; private set; } + public long Uid { get; private set; } /// /// TBD @@ -844,7 +844,7 @@ public static Props ReliableDeliverySupervisorProps( AkkaProtocolHandle handleOrActive, Address localAddress, Address remoteAddress, - int? refuseUid, + long? refuseUid, AkkaProtocolTransport transport, RemoteSettings settings, AkkaPduCodec codec, @@ -1022,7 +1022,7 @@ public EndpointWriter( AkkaProtocolHandle handleOrActive, Address localAddress, Address remoteAddress, - int? refuseUid, + long? refuseUid, AkkaProtocolTransport transport, RemoteSettings settings, AkkaPduCodec codec, @@ -1055,7 +1055,7 @@ public EndpointWriter( private readonly ILoggingAdapter _log = Context.GetLogger(); private readonly Information _transportInformation; - private readonly int? _refuseUid; + private readonly long? _refuseUid; private readonly AkkaPduCodec _codec; private readonly IActorRef _reliableDeliverySupervisor; private readonly ExtendedActorSystem _system; @@ -1323,7 +1323,7 @@ private IActorRef StartReadEndpoint(AkkaProtocolHandle handle) Context.ActorOf(RARP.For(Context.System) .ConfigureDispatcher( EndpointReader.ReaderProps(LocalAddress, RemoteAddress, Transport, Settings, _codec, _msgDispatcher, - Inbound, (int)handle.HandshakeInfo.Uid, _receiveBuffers, _reliableDeliverySupervisor) + Inbound, handle.HandshakeInfo.Uid, _receiveBuffers, _reliableDeliverySupervisor) .WithDeploy(Deploy.Local)), $"endpointReader-{AddressUrlEncoder.Encode(RemoteAddress)}-{_readerId.Next()}"); Context.Watch(newReader); @@ -1647,7 +1647,7 @@ bool WritePrioLoop() private const int MaxWriteCount = 50; public static Props EndpointWriterProps(AkkaProtocolHandle handleOrActive, Address localAddress, - Address remoteAddress, int? refuseUid, AkkaProtocolTransport transport, RemoteSettings settings, + Address remoteAddress, long? refuseUid, AkkaProtocolTransport transport, RemoteSettings settings, AkkaPduCodec codec, ConcurrentDictionary receiveBuffers, IActorRef reliableDeliverySupervisor = null) { return Props.Create( @@ -1778,7 +1778,7 @@ public EndpointReader( AkkaPduCodec codec, IInboundMessageDispatcher msgDispatch, bool inbound, - int uid, + long uid, ConcurrentDictionary receiveBuffers, IActorRef reliableDeliverySupervisor = null) : base(localAddress, remoteAddress, transport, settings) @@ -1797,7 +1797,7 @@ public EndpointReader( private readonly AkkaPduCodec _codec; private readonly IActorRef _reliableDeliverySupervisor; private readonly ConcurrentDictionary _receiveBuffers; - private readonly int _uid; + private readonly long _uid; private readonly IInboundMessageDispatcher _msgDispatch; private readonly IRemoteActorRefProvider _provider; @@ -1994,7 +1994,7 @@ public static Props ReaderProps( AkkaPduCodec codec, IInboundMessageDispatcher dispatcher, bool inbound, - int uid, + long uid, ConcurrentDictionary receiveBuffers, IActorRef reliableDeliverySupervisor = null) { diff --git a/src/core/Akka.Remote/EndpointManager.cs b/src/core/Akka.Remote/EndpointManager.cs index 2444e99879b..0797836fcf2 100644 --- a/src/core/Akka.Remote/EndpointManager.cs +++ b/src/core/Akka.Remote/EndpointManager.cs @@ -55,7 +55,7 @@ public sealed class Pass : EndpointPolicy /// /// TBD /// TBD - public Pass(IActorRef endpoint, int? uid) + public Pass(IActorRef endpoint, long? uid) : base(false) { Uid = uid; @@ -70,7 +70,7 @@ public Pass(IActorRef endpoint, int? uid) /// /// The endpoint UID, if it's currently known /// - public int? Uid { get; } + public long? Uid { get; } } /// @@ -93,14 +93,14 @@ public Gated(Deadline deadline) /// public sealed class Quarantined : EndpointPolicy { - public Quarantined(int uid, Deadline deadline) + public Quarantined(long uid, Deadline deadline) : base(true) { Uid = uid; Deadline = deadline; } - public int Uid { get; } + public long Uid { get; } public Deadline Deadline { get; } } @@ -217,7 +217,7 @@ public sealed class Quarantine : RemotingCommand /// /// TBD /// TBD - public Quarantine(Address remoteAddress, int? uid) + public Quarantine(Address remoteAddress, long? uid) { Uid = uid; RemoteAddress = remoteAddress; @@ -231,7 +231,7 @@ public Quarantine(Address remoteAddress, int? uid) /// /// TBD /// - public int? Uid { get; private set; } + public long? Uid { get; private set; } } /// @@ -399,7 +399,7 @@ public sealed class ResendState : IEquatable /// /// TBD /// TBD - public ResendState(int uid, AckedReceiveBuffer buffer) + public ResendState(long uid, AckedReceiveBuffer buffer) { Buffer = buffer; Uid = uid; @@ -408,7 +408,7 @@ public ResendState(int uid, AckedReceiveBuffer buffer) /// /// TBD /// - public int Uid { get; private set; } + public long Uid { get; private set; } /// /// TBD @@ -431,7 +431,7 @@ public override int GetHashCode() { unchecked { - return (Uid * 397) ^ Buffer.GetHashCode(); + return (Uid.GetHashCode() * 397) ^ Buffer.GetHashCode(); } } } diff --git a/src/core/Akka.Remote/EndpointRegistry.cs b/src/core/Akka.Remote/EndpointRegistry.cs index a144c2ce4bd..5aff1eea609 100644 --- a/src/core/Akka.Remote/EndpointRegistry.cs +++ b/src/core/Akka.Remote/EndpointRegistry.cs @@ -18,8 +18,8 @@ namespace Akka.Remote /// internal sealed class EndpointRegistry { - private Dictionary _addressToRefuseUid = new(); - private readonly Dictionary _addressToReadonly = new(); + private Dictionary _addressToRefuseUid = new(); + private readonly Dictionary _addressToReadonly = new(); private Dictionary _addressToWritable = new(); @@ -38,7 +38,7 @@ internal sealed class EndpointRegistry /// in the registry. /// /// The actor reference. - public IActorRef RegisterWritableEndpoint(Address address, IActorRef endpoint, int? uid) + public IActorRef RegisterWritableEndpoint(Address address, IActorRef endpoint, long? uid) { if (_addressToWritable.TryGetValue(address, out var existing)) { @@ -60,7 +60,7 @@ public IActorRef RegisterWritableEndpoint(Address address, IActorRef endpoint, i /// /// The address of the remote system. /// The UID of the remote system. - public void RegisterWritableEndpointUid(Address remoteAddress, int uid) + public void RegisterWritableEndpointUid(Address remoteAddress, long uid) { if (_addressToWritable.TryGetValue(remoteAddress, out var existing)) { @@ -76,7 +76,7 @@ public void RegisterWritableEndpointUid(Address remoteAddress, int uid) /// The remote address of the quarantined system. /// The refused UID of the remote system. /// The timeframe for releasing quarantine. - public void RegisterWritableEndpointRefuseUid(Address remoteAddress, int refuseUid, Deadline timeOfRelease) + public void RegisterWritableEndpointRefuseUid(Address remoteAddress, long refuseUid, Deadline timeOfRelease) { _addressToRefuseUid[remoteAddress] = (refuseUid, timeOfRelease); } @@ -88,7 +88,7 @@ public void RegisterWritableEndpointRefuseUid(Address remoteAddress, int refuseU /// The local endpoint actor who owns this connection. /// The UID of the remote actor system. Can be null. /// The actor reference. - public IActorRef RegisterReadOnlyEndpoint(Address address, IActorRef endpoint, int uid) + public IActorRef RegisterReadOnlyEndpoint(Address address, IActorRef endpoint, long uid) { _addressToReadonly[address] = (endpoint, uid); _readonlyToAddress[endpoint] = address; @@ -138,7 +138,7 @@ public Address AddressForWriter(IActorRef writer) /// /// The remote address to check. /// A tuple containing the actor reference and the remote system UID, if they exist. Otherwise null. - public (IActorRef, int)? ReadOnlyEndpointFor(Address address) + public (IActorRef, long)? ReadOnlyEndpointFor(Address address) { if (!_addressToReadonly.TryGetValue(address, out var tmp)) return null; @@ -172,7 +172,7 @@ public bool IsReadOnly(IActorRef endpoint) /// The address to check. /// The current UID of . /// true if this system is under quarantine with its current UID, false otherwise. - public bool IsQuarantined(Address address, int uid) + public bool IsQuarantined(Address address, long uid) { // timeOfRelease is only used for garbage collection. If an address is still probed, we should report the // known fact that it is quarantined. @@ -196,7 +196,7 @@ public bool IsQuarantined(Address address, int uid) /// /// The address to check. /// The refused UID if one exists for this address; otherwise null. - public int? RefuseUid(Address address) + public long? RefuseUid(Address address) { // timeOfRelease is only used for garbage collection. If an address is still probed, we should report the // known fact that it is quarantined. @@ -280,7 +280,7 @@ public void MarkAsFailed(IActorRef endpoint, Deadline timeOfRelease) /// The address to quarantine. /// The UID of the current address. /// The timeframe to release the quarantine. - public void MarkAsQuarantined(Address address, int uid, Deadline timeOfRelease) + public void MarkAsQuarantined(Address address, long uid, Deadline timeOfRelease) { _addressToWritable[address] = new EndpointManager.Quarantined(uid, timeOfRelease); _addressToRefuseUid[address] = (uid, timeOfRelease); diff --git a/src/core/Akka.Remote/RemoteActorRefProvider.cs b/src/core/Akka.Remote/RemoteActorRefProvider.cs index 8401f28c89a..fa0b1f5b6e9 100644 --- a/src/core/Akka.Remote/RemoteActorRefProvider.cs +++ b/src/core/Akka.Remote/RemoteActorRefProvider.cs @@ -98,7 +98,7 @@ public interface IRemoteActorRefProvider : IActorRefProvider /// UID of the remote system, if the uid is not defined it will not be a strong quarantine but /// the current endpoint writer will be stopped (dropping system messages) and the address will be gated /// - void Quarantine(Address address, int? uid); + void Quarantine(Address address, long? uid); } /// @@ -634,7 +634,7 @@ public void UseActorOnNode(RemoteActorRef actor, Props props, Deploy deploy, IIn /// UID of the remote system, if the uid is not defined it will not be a strong quarantine but /// the current endpoint writer will be stopped (dropping system messages) and the address will be gated /// - public void Quarantine(Address address, int? uid) + public void Quarantine(Address address, long? uid) { Transport.Quarantine(address, uid); } diff --git a/src/core/Akka.Remote/RemoteTransport.cs b/src/core/Akka.Remote/RemoteTransport.cs index 65eca5d9b6c..9baa1090761 100644 --- a/src/core/Akka.Remote/RemoteTransport.cs +++ b/src/core/Akka.Remote/RemoteTransport.cs @@ -115,7 +115,7 @@ protected RemoteTransport(ExtendedActorSystem system, RemoteActorRefProvider pro /// Address of the remote system to be quarantined /// UID of the remote system; if the uid is not defined it will not be a strong quarantine but the current /// endpoint writer will be stopped (dropping system messages) and the address will be gated. - public abstract void Quarantine(Address address, int? uid); + public abstract void Quarantine(Address address, long? uid); } /// diff --git a/src/core/Akka.Remote/RemoteWatcher.cs b/src/core/Akka.Remote/RemoteWatcher.cs index 2d613fe1ef4..3097df4ccc8 100644 --- a/src/core/Akka.Remote/RemoteWatcher.cs +++ b/src/core/Akka.Remote/RemoteWatcher.cs @@ -140,13 +140,13 @@ private Heartbeat() /// public class HeartbeatRsp : IPriorityMessage { - readonly int _addressUid; + readonly long _addressUid; /// /// TBD /// /// TBD - public HeartbeatRsp(int addressUid) + public HeartbeatRsp(long addressUid) { _addressUid = addressUid; } @@ -154,7 +154,7 @@ public HeartbeatRsp(int addressUid) /// /// TBD /// - public int AddressUid + public long AddressUid { get { return _addressUid; } } @@ -387,7 +387,7 @@ TimeSpan heartbeatExpectedResponseAfter /// protected HashSet
Unreachable { get; } = new(); - private readonly Dictionary _addressUids = new(); + private readonly Dictionary _addressUids = new(); private readonly ICancelable _heartbeatCancelable; private readonly ICancelable _failureDetectorReaperCancelable; @@ -465,7 +465,7 @@ private void ReceiveHeartbeat() Sender.Tell(_selfHeartbeatRspMsg); } - private void ReceiveHeartbeatRsp(int uid) + private void ReceiveHeartbeatRsp(long uid) { var from = Sender.Path.Address; @@ -476,7 +476,7 @@ private void ReceiveHeartbeatRsp(int uid) if (WatcheeByNodes.ContainsKey(from) && !Unreachable.Contains(from)) { - if (!_addressUids.TryGetValue(from, out int addressUid) || addressUid != uid) + if (!_addressUids.TryGetValue(from, out long addressUid) || addressUid != uid) ReWatch(from); _addressUids[from] = uid; @@ -492,7 +492,7 @@ private void ReapUnreachable() { Log.Warning("Detected unreachable: [{0}]", a); var nullableAddressUid = - _addressUids.TryGetValue(a, out int addressUid) ? new int?(addressUid) : null; + _addressUids.TryGetValue(a, out long addressUid) ? new long?(addressUid) : null; Quarantine(a, nullableAddressUid); PublishAddressTerminated(a); @@ -515,7 +515,7 @@ protected virtual void PublishAddressTerminated(Address address) /// /// TBD /// TBD - protected virtual void Quarantine(Address address, int? addressUid) + protected virtual void Quarantine(Address address, long? addressUid) { _remoteProvider.Quarantine(address, addressUid); } diff --git a/src/core/Akka.Remote/Remoting.cs b/src/core/Akka.Remote/Remoting.cs index d1c06559338..bbd3146a2c3 100644 --- a/src/core/Akka.Remote/Remoting.cs +++ b/src/core/Akka.Remote/Remoting.cs @@ -337,7 +337,7 @@ public override Address LocalAddressForRemote(Address remote) /// /// This exception is thrown when trying to quarantine a system but remoting is not running. /// - public override void Quarantine(Address address, int? uid) + public override void Quarantine(Address address, long? uid) { if (_endpointManager == null) { diff --git a/src/core/Akka.Remote/RemotingLifecycleEvent.cs b/src/core/Akka.Remote/RemotingLifecycleEvent.cs index f21b61dc814..6b3afb3fa2b 100644 --- a/src/core/Akka.Remote/RemotingLifecycleEvent.cs +++ b/src/core/Akka.Remote/RemotingLifecycleEvent.cs @@ -305,7 +305,7 @@ public sealed class QuarantinedEvent : RemotingLifecycleEvent /// /// TBD /// TBD - public QuarantinedEvent(Address address, int uid) + public QuarantinedEvent(Address address, long uid) { Uid = uid; Address = address; @@ -319,7 +319,7 @@ public QuarantinedEvent(Address address, int uid) /// /// TBD /// - public int Uid { get; private set; } + public long Uid { get; private set; } /// /// TBD diff --git a/src/core/Akka.Remote/Serialization/MiscMessageSerializer.cs b/src/core/Akka.Remote/Serialization/MiscMessageSerializer.cs index c75bddcde28..d88a9473eb1 100644 --- a/src/core/Akka.Remote/Serialization/MiscMessageSerializer.cs +++ b/src/core/Akka.Remote/Serialization/MiscMessageSerializer.cs @@ -283,14 +283,14 @@ private IActorRef ActorRefFromProto(byte[] bytes) private byte[] HeartbeatRspToProto(RemoteWatcher.HeartbeatRsp heartbeatRsp) { var message = new Proto.Msg.RemoteWatcherHeartbeatResponse(); - message.Uid = (ulong)heartbeatRsp.AddressUid; // TODO: change to uint32 + message.Uid = (ulong)heartbeatRsp.AddressUid; return message.ToByteArray(); } private RemoteWatcher.HeartbeatRsp HearthbeatRspFromProto(byte[] bytes) { var message = Proto.Msg.RemoteWatcherHeartbeatResponse.Parser.ParseFrom(bytes); - return new RemoteWatcher.HeartbeatRsp((int)message.Uid); + return new RemoteWatcher.HeartbeatRsp((long)message.Uid); } // diff --git a/src/core/Akka.Remote/Transport/AkkaPduCodec.cs b/src/core/Akka.Remote/Transport/AkkaPduCodec.cs index c4d54c81475..fbce8484774 100644 --- a/src/core/Akka.Remote/Transport/AkkaPduCodec.cs +++ b/src/core/Akka.Remote/Transport/AkkaPduCodec.cs @@ -882,7 +882,7 @@ private IAkkaPdu DecodeControlPdu(AkkaControlMessage controlPdu) var handshakeInfo = controlPdu.HandshakeInfo; if (handshakeInfo != null) // HasHandshakeInfo { - return new Associate(new HandshakeInfo(DecodeAddress(handshakeInfo.Origin), (int)handshakeInfo.Uid)); + return new Associate(new HandshakeInfo(DecodeAddress(handshakeInfo.Origin), (long)handshakeInfo.Uid)); } break; case CommandType.Disassociate: diff --git a/src/core/Akka.Remote/Transport/AkkaProtocolTransport.cs b/src/core/Akka.Remote/Transport/AkkaProtocolTransport.cs index f554221177f..f7019b729d4 100644 --- a/src/core/Akka.Remote/Transport/AkkaProtocolTransport.cs +++ b/src/core/Akka.Remote/Transport/AkkaProtocolTransport.cs @@ -161,7 +161,7 @@ public override Task ManagementCommand(object message) /// TBD /// TBD /// TBD - public async Task Associate(Address remoteAddress, int? refuseUid) + public async Task Associate(Address remoteAddress, long? refuseUid) { // Prepare a Task and pass its completion source to the manager var statusPromise = new TaskCompletionSource(); @@ -261,7 +261,7 @@ private string ActorNameFor(Address remoteAddress) } private void CreateOutboundStateActor(Address remoteAddress, - TaskCompletionSource statusPromise, int? refuseUid) + TaskCompletionSource statusPromise, long? refuseUid) { var stateActorLocalAddress = LocalAddress; var stateActorSettings = _settings; @@ -298,7 +298,7 @@ internal sealed class AssociateUnderlyingRefuseUid : INoSerializationVerificatio /// TBD /// TBD /// TBD - public AssociateUnderlyingRefuseUid(Address remoteAddress, TaskCompletionSource statusCompletionSource, int? refuseUid = null) + public AssociateUnderlyingRefuseUid(Address remoteAddress, TaskCompletionSource statusCompletionSource, long? refuseUid = null) { RefuseUid = refuseUid; StatusCompletionSource = statusCompletionSource; @@ -318,7 +318,7 @@ public AssociateUnderlyingRefuseUid(Address remoteAddress, TaskCompletionSource< /// /// TBD /// - public int? RefuseUid { get; private set; } + public long? RefuseUid { get; private set; } } /// @@ -331,7 +331,7 @@ internal sealed class HandshakeInfo /// /// TBD /// TBD - public HandshakeInfo(Address origin, int uid) + public HandshakeInfo(Address origin, long uid) { Origin = origin; Uid = uid; @@ -345,7 +345,7 @@ public HandshakeInfo(Address origin, int uid) /// /// TBD /// - public int Uid { get; private set; } + public long Uid { get; private set; } /// public override bool Equals(object obj) @@ -725,7 +725,7 @@ internal sealed class ProtocolStateActor : FSMTBD public ProtocolStateActor(HandshakeInfo handshakeInfo, Address remoteAddress, TaskCompletionSource statusCompletionSource, Transport transport, - AkkaProtocolSettings settings, AkkaPduCodec codec, FailureDetector failureDetector, int? refuseUid = null) + AkkaProtocolSettings settings, AkkaPduCodec codec, FailureDetector failureDetector, long? refuseUid = null) : this( new OutboundUnassociated(remoteAddress, statusCompletionSource, transport), handshakeInfo, settings, codec, failureDetector, refuseUid) @@ -811,7 +811,7 @@ public ProtocolStateActor(HandshakeInfo handshakeInfo, AssociationHandle wrapped /// /// /// - private ProtocolStateActor(InitialProtocolStateData initialData, HandshakeInfo localHandshakeInfo, AkkaProtocolSettings settings, AkkaPduCodec codec, FailureDetector failureDetector, int? refuseUid) + private ProtocolStateActor(InitialProtocolStateData initialData, HandshakeInfo localHandshakeInfo, AkkaProtocolSettings settings, AkkaPduCodec codec, FailureDetector failureDetector, long? refuseUid) { _initialData = initialData; _localHandshakeInfo = localHandshakeInfo; @@ -1386,7 +1386,7 @@ private void PublishError(UnderlyingTransportError transportError) /// TBD public static Props OutboundProps(HandshakeInfo handshakeInfo, Address remoteAddress, TaskCompletionSource statusCompletionSource, - Transport transport, AkkaProtocolSettings settings, AkkaPduCodec codec, FailureDetector failureDetector, int? refuseUid = null) + Transport transport, AkkaProtocolSettings settings, AkkaPduCodec codec, FailureDetector failureDetector, long? refuseUid = null) { return Props.Create(() => new ProtocolStateActor(handshakeInfo, remoteAddress, statusCompletionSource, transport, settings, codec, failureDetector, refuseUid)); } From 6044662d5ad623261d6c47c59d25ea61c994ca0b Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 3 Jul 2026 17:56:35 -0500 Subject: [PATCH 2/9] widen-system-uid task 6.3: pin 64-bit system uid constraint in serializer-v2 schema (Decision 11) --- openspec/changes/serializer-v2/design.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/openspec/changes/serializer-v2/design.md b/openspec/changes/serializer-v2/design.md index 94db97a4fc4..7f9a8a37d11 100644 --- a/openspec/changes/serializer-v2/design.md +++ b/openspec/changes/serializer-v2/design.md @@ -105,6 +105,12 @@ Source-generated MessagePack serializers should be implemented only after this f Rationale: sourcegen validates the V2 API through real serialization, classic remoting, and persistence paths before Artery envelopes depend on it. +### 11. System UID Is 64-Bit In Any V2 Schema + +Any V2 or sourcegen schema that carries the address/system UID (`UniqueAddress`, quarantine, heartbeat, handshake, Artery envelope origin) MUST emit and read it as a 64-bit integer (`long`). No V2 schema may introduce a 32-bit uid field. + +Rationale: `widen-system-uid-to-64bit` (Milestone 3.5) re-types the system UID to `long` end-to-end as a prerequisite for Artery, whose frame header carries a 64-bit origin UID. This change does not block on serializer-v2, but constrains its schema (see that change's design.md, Decision 4). + ## Risks / Trade-offs **Compatibility inheritance tension**: `SerializerV2` being usable as `Serializer` keeps this PR compatible, but permits awkward compositions such as wrapping V2 with `SerializerV1Adapter`. Guardrails should be added before native V2 serializers become common. From 259a94e05f379a15be4f35c022ada88b5332372c Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 3 Jul 2026 18:07:45 -0500 Subject: [PATCH 3/9] widen-system-uid task groups 2,6.1,6.2: Akka.Cluster + DData int->long system uid - UniqueAddress(Address, long uid) + long Uid; GetHashCode via Uid.GetHashCode() (value-identical to the old int hash for legacy-range uids); CompareTo/Equals already long-safe - ClusterMessages.proto UniqueAddress.uid uint32 -> uint64 (same varint wire type; binary-compatible for uids <= uint32.MaxValue; regenerated by Grpc.Tools) - ClusterMessageSerializer: (uint)/(int) narrowing replaced with lossless (ulong)/(long) reinterpret casts at the proto boundary - DData SerializationSupport.UniqueAddressFromProto: (int) narrowing removed (ReplicatorMessages.proto already int64) - vclock node name verified string-only (ClusterDaemon.VclockName) - unchanged Whole solution builds green. --- openspec/changes/widen-system-uid-to-64bit/tasks.md | 10 +++++----- .../Serialization/SerializationSupport.cs | 2 +- src/core/Akka.Cluster/Member.cs | 6 +++--- .../Serialization/ClusterMessageSerializer.cs | 4 ++-- src/protobuf/ClusterMessages.proto | 2 +- 5 files changed, 12 insertions(+), 12 deletions(-) diff --git a/openspec/changes/widen-system-uid-to-64bit/tasks.md b/openspec/changes/widen-system-uid-to-64bit/tasks.md index 2f4bddfaf21..327ad798336 100644 --- a/openspec/changes/widen-system-uid-to-64bit/tasks.md +++ b/openspec/changes/widen-system-uid-to-64bit/tasks.md @@ -6,9 +6,9 @@ ## 2. Cluster -- [ ] 2.1 `UniqueAddress(Address, long uid)` + `long Uid`; fix identity/sort/hash (`Member`) -- [ ] 2.2 Verify gossip vector-clock node name (`VclockName` = address + "-" + uid decimal string) — behavior unchanged -- [ ] 2.3 `ClusterDaemon` `Quarantined(new UniqueAddress(addr, long uid))` +- [x] 2.1 `UniqueAddress(Address, long uid)` + `long Uid`; fix identity/sort/hash (`Member`) +- [x] 2.2 Verify gossip vector-clock node name (`VclockName` = address + "-" + uid decimal string) — behavior unchanged (string-only, `ClusterDaemon.VclockName`; hash also value-identical for int-range uids) +- [x] 2.3 `ClusterDaemon` `Quarantined(new UniqueAddress(addr, long uid))` ## 3. Remoting state machine (internal `int → long`) @@ -30,8 +30,8 @@ ## 6. Wire + serializers -- [ ] 6.1 `ClusterMessages.proto` `UniqueAddress.uid` `uint32 → uint64`; regenerate -- [ ] 6.2 Remove narrowing casts: `ClusterMessageSerializer`, `MiscMessageSerializer`, `AkkaPduCodec`, DData `SerializationSupport.UniqueAddressFromProto` +- [x] 6.1 `ClusterMessages.proto` `UniqueAddress.uid` `uint32 → uint64`; regenerate (Grpc.Tools at build time) +- [x] 6.2 Remove narrowing casts: `ClusterMessageSerializer`, `MiscMessageSerializer`, `AkkaPduCodec`, DData `SerializationSupport.UniqueAddressFromProto` - [x] 6.3 Coordinate `serializer-v2` schema to emit/read 64-bit uid (Decision 11 in serializer-v2 design.md) ## 7. API approval + build diff --git a/src/contrib/cluster/Akka.DistributedData/Serialization/SerializationSupport.cs b/src/contrib/cluster/Akka.DistributedData/Serialization/SerializationSupport.cs index db5280f6bd6..3612f9c9633 100644 --- a/src/contrib/cluster/Akka.DistributedData/Serialization/SerializationSupport.cs +++ b/src/contrib/cluster/Akka.DistributedData/Serialization/SerializationSupport.cs @@ -142,7 +142,7 @@ public static Proto.Msg.UniqueAddress UniqueAddressToProto(UniqueAddress address public UniqueAddress UniqueAddressFromProto(Proto.Msg.UniqueAddress address) { - return new UniqueAddress(AddressFromProto(address.Address), (int)address.Uid); + return new UniqueAddress(AddressFromProto(address.Address), address.Uid); } public static Proto.Msg.VersionVector VersionVectorToProto(VersionVector versionVector) diff --git a/src/core/Akka.Cluster/Member.cs b/src/core/Akka.Cluster/Member.cs index 36780edbf2f..90d3cfc6835 100644 --- a/src/core/Akka.Cluster/Member.cs +++ b/src/core/Akka.Cluster/Member.cs @@ -474,14 +474,14 @@ public class UniqueAddress : IComparable, IEquatable /// A random long integer used to signal the incarnation of this cluster instance. /// - public int Uid { get; } + public long Uid { get; } /// /// Creates a new unique address instance. /// /// The original Akka /// The UID for the cluster instance. - public UniqueAddress(Address address, int uid) + public UniqueAddress(Address address, long uid) { Uid = uid; Address = address; @@ -506,7 +506,7 @@ public bool Equals(UniqueAddress other) public override int GetHashCode() { - return Uid; + return Uid.GetHashCode(); } /// diff --git a/src/core/Akka.Cluster/Serialization/ClusterMessageSerializer.cs b/src/core/Akka.Cluster/Serialization/ClusterMessageSerializer.cs index f857e5b96da..ad4ece50190 100644 --- a/src/core/Akka.Cluster/Serialization/ClusterMessageSerializer.cs +++ b/src/core/Akka.Cluster/Serialization/ClusterMessageSerializer.cs @@ -578,14 +578,14 @@ internal static Proto.Msg.UniqueAddress UniqueAddressToProto(UniqueAddress uniqu { var message = new Proto.Msg.UniqueAddress(); message.Address = AddressToProto(uniqueAddress.Address); - message.Uid = (uint)uniqueAddress.Uid; + message.Uid = (ulong)uniqueAddress.Uid; return message; } [MethodImpl(MethodImplOptions.AggressiveInlining)] internal static UniqueAddress UniqueAddressFrom(Proto.Msg.UniqueAddress uniqueAddressProto) { - return new UniqueAddress(AddressFrom(uniqueAddressProto.Address), (int)uniqueAddressProto.Uid); + return new UniqueAddress(AddressFrom(uniqueAddressProto.Address), (long)uniqueAddressProto.Uid); } [MethodImpl(MethodImplOptions.AggressiveInlining)] diff --git a/src/protobuf/ClusterMessages.proto b/src/protobuf/ClusterMessages.proto index 4206c0f3c21..81dd452e659 100644 --- a/src/protobuf/ClusterMessages.proto +++ b/src/protobuf/ClusterMessages.proto @@ -173,7 +173,7 @@ message VectorClock { // Defines a remote address with uid. message UniqueAddress { Akka.Remote.Serialization.Proto.Msg.AddressData address = 1; - uint32 uid = 2; + uint64 uid = 2; } /**************************************** From 25f7c56a1114fb6432a368ded4dd7b2c32c1811f Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 3 Jul 2026 18:11:42 -0500 Subject: [PATCH 4/9] widen-system-uid tasks 7.1-7.2: API approval files for the int->long hard break ApproveRemote + ApproveCluster (.DotNet and .Net variants): the ~10 inventoried members re-typed int->long, nothing else. All 18 API approval tests pass; dotnet build -warnaserror clean on the full solution. --- .../widen-system-uid-to-64bit/tasks.md | 4 ++-- ...APISpec.ApproveCluster.DotNet.verified.txt | 4 ++-- ...oreAPISpec.ApproveCluster.Net.verified.txt | 4 ++-- ...eAPISpec.ApproveRemote.DotNet.verified.txt | 20 +++++++++---------- ...CoreAPISpec.ApproveRemote.Net.verified.txt | 20 +++++++++---------- 5 files changed, 26 insertions(+), 26 deletions(-) diff --git a/openspec/changes/widen-system-uid-to-64bit/tasks.md b/openspec/changes/widen-system-uid-to-64bit/tasks.md index 327ad798336..e3f1a90a508 100644 --- a/openspec/changes/widen-system-uid-to-64bit/tasks.md +++ b/openspec/changes/widen-system-uid-to-64bit/tasks.md @@ -36,8 +36,8 @@ ## 7. API approval + build -- [ ] 7.1 Update `Akka.API.Tests` approved files (`ApproveRemote.*`, `ApproveCluster.*`) for the hard break -- [ ] 7.2 `dotnet build -warnaserror` clean +- [x] 7.1 Update `Akka.API.Tests` approved files (`ApproveRemote.*`, `ApproveCluster.*`) for the hard break +- [x] 7.2 `dotnet build -warnaserror` clean - [ ] 7.3 Akka.Remote + Akka.Cluster + DistributedData tests green ## 8. Compatibility + docs diff --git a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCluster.DotNet.verified.txt b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCluster.DotNet.verified.txt index 16a46e94145..af790c4801f 100644 --- a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCluster.DotNet.verified.txt +++ b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCluster.DotNet.verified.txt @@ -285,9 +285,9 @@ namespace Akka.Cluster } public class UniqueAddress : System.IComparable, System.IComparable, System.IEquatable { - public UniqueAddress(Akka.Actor.Address address, int uid) { } + public UniqueAddress(Akka.Actor.Address address, long uid) { } public Akka.Actor.Address Address { get; } - public int Uid { get; } + public long Uid { get; } public int CompareTo(Akka.Cluster.UniqueAddress uniqueAddress) { } public bool Equals(Akka.Cluster.UniqueAddress other) { } public override bool Equals(object obj) { } diff --git a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCluster.Net.verified.txt b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCluster.Net.verified.txt index 53d3bdc0255..42b4e03e0a0 100644 --- a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCluster.Net.verified.txt +++ b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveCluster.Net.verified.txt @@ -285,9 +285,9 @@ namespace Akka.Cluster } public class UniqueAddress : System.IComparable, System.IComparable, System.IEquatable { - public UniqueAddress(Akka.Actor.Address address, int uid) { } + public UniqueAddress(Akka.Actor.Address address, long uid) { } public Akka.Actor.Address Address { get; } - public int Uid { get; } + public long Uid { get; } public int CompareTo(Akka.Cluster.UniqueAddress uniqueAddress) { } public bool Equals(Akka.Cluster.UniqueAddress other) { } public override bool Equals(object obj) { } diff --git a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveRemote.DotNet.verified.txt b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveRemote.DotNet.verified.txt index a9c2c8adbce..e20f156893a 100644 --- a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveRemote.DotNet.verified.txt +++ b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveRemote.DotNet.verified.txt @@ -23,14 +23,14 @@ namespace Akka.Remote { public class AddressUid : Akka.Actor.IExtension { - public readonly int Uid; + public readonly long Uid; public AddressUid() { } } public class AddressUidExtension : Akka.Actor.ExtensionIdProvider { public AddressUidExtension() { } public override Akka.Remote.AddressUid CreateExtension(Akka.Actor.ExtendedActorSystem system) { } - public static int Uid(Akka.Actor.ActorSystem system) { } + public static long Uid(Akka.Actor.ActorSystem system) { } } public class AkkaProtocolSettings { @@ -143,7 +143,7 @@ namespace Akka.Remote bool HasAddress(Akka.Actor.Address address); Akka.Actor.IActorRef InternalResolveActorRef(string path); Akka.Actor.Deploy LookUpRemotes(System.Collections.Generic.IEnumerable p); - void Quarantine(Akka.Actor.Address address, int? uid); + void Quarantine(Akka.Actor.Address address, long? uid); Akka.Actor.IInternalActorRef ResolveActorRefWithLocalAddress(string path, Akka.Actor.Address localAddress); void UseActorOnNode(Akka.Remote.RemoteActorRef actor, Akka.Actor.Props props, Akka.Actor.Deploy deploy, Akka.Actor.IInternalActorRef supervisor); } @@ -161,9 +161,9 @@ namespace Akka.Remote } public sealed class QuarantinedEvent : Akka.Remote.RemotingLifecycleEvent { - public QuarantinedEvent(Akka.Actor.Address address, int uid) { } + public QuarantinedEvent(Akka.Actor.Address address, long uid) { } public Akka.Actor.Address Address { get; } - public int Uid { get; } + public long Uid { get; } public override Akka.Event.LogLevel LogLevel() { } public override string ToString() { } } @@ -219,7 +219,7 @@ namespace Akka.Remote public virtual void Init(Akka.Actor.Internal.ActorSystemImpl system) { } public Akka.Actor.IActorRef InternalResolveActorRef(string path) { } public Akka.Actor.Deploy LookUpRemotes(System.Collections.Generic.IEnumerable p) { } - public void Quarantine(Akka.Actor.Address address, int? uid) { } + public void Quarantine(Akka.Actor.Address address, long? uid) { } public void RegisterTempActor(Akka.Actor.IInternalActorRef actorRef, Akka.Actor.ActorPath path) { } public Akka.Actor.IActorRef ResolveActorRef(Akka.Actor.ActorPath actorPath) { } public Akka.Actor.IActorRef ResolveActorRef(string path) { } @@ -286,7 +286,7 @@ namespace Akka.Remote public abstract Akka.Actor.Address LocalAddressForRemote(Akka.Actor.Address remote); public abstract System.Threading.Tasks.Task ManagementCommand(object cmd); public abstract System.Threading.Tasks.Task ManagementCommand(object cmd, System.Threading.CancellationToken cancellationToken); - public abstract void Quarantine(Akka.Actor.Address address, int? uid); + public abstract void Quarantine(Akka.Actor.Address address, long? uid); public abstract void Send(object message, Akka.Actor.IActorRef sender, Akka.Remote.RemoteActorRef recipient); public abstract System.Threading.Tasks.Task Shutdown(); public abstract void Start(); @@ -308,7 +308,7 @@ namespace Akka.Remote protected override void OnReceive(object message) { } protected override void PostStop() { } protected virtual void PublishAddressTerminated(Akka.Actor.Address address) { } - protected virtual void Quarantine(Akka.Actor.Address address, int? addressUid) { } + protected virtual void Quarantine(Akka.Actor.Address address, long? addressUid) { } protected void RemoveWatch(Akka.Actor.IInternalActorRef watchee, Akka.Actor.IInternalActorRef watcher) { } protected void RemoveWatchee(Akka.Actor.IInternalActorRef watchee) { } protected void UnwatchNode(Akka.Actor.Address watcheeAddress) { } @@ -325,8 +325,8 @@ namespace Akka.Remote } public class HeartbeatRsp { - public HeartbeatRsp(int addressUid) { } - public int AddressUid { get; } + public HeartbeatRsp(long addressUid) { } + public long AddressUid { get; } } public class HeartbeatTick { diff --git a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveRemote.Net.verified.txt b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveRemote.Net.verified.txt index dca096bb186..5637795f550 100644 --- a/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveRemote.Net.verified.txt +++ b/src/core/Akka.API.Tests/verify/CoreAPISpec.ApproveRemote.Net.verified.txt @@ -23,14 +23,14 @@ namespace Akka.Remote { public class AddressUid : Akka.Actor.IExtension { - public readonly int Uid; + public readonly long Uid; public AddressUid() { } } public class AddressUidExtension : Akka.Actor.ExtensionIdProvider { public AddressUidExtension() { } public override Akka.Remote.AddressUid CreateExtension(Akka.Actor.ExtendedActorSystem system) { } - public static int Uid(Akka.Actor.ActorSystem system) { } + public static long Uid(Akka.Actor.ActorSystem system) { } } public class AkkaProtocolSettings { @@ -143,7 +143,7 @@ namespace Akka.Remote bool HasAddress(Akka.Actor.Address address); Akka.Actor.IActorRef InternalResolveActorRef(string path); Akka.Actor.Deploy LookUpRemotes(System.Collections.Generic.IEnumerable p); - void Quarantine(Akka.Actor.Address address, int? uid); + void Quarantine(Akka.Actor.Address address, long? uid); Akka.Actor.IInternalActorRef ResolveActorRefWithLocalAddress(string path, Akka.Actor.Address localAddress); void UseActorOnNode(Akka.Remote.RemoteActorRef actor, Akka.Actor.Props props, Akka.Actor.Deploy deploy, Akka.Actor.IInternalActorRef supervisor); } @@ -161,9 +161,9 @@ namespace Akka.Remote } public sealed class QuarantinedEvent : Akka.Remote.RemotingLifecycleEvent { - public QuarantinedEvent(Akka.Actor.Address address, int uid) { } + public QuarantinedEvent(Akka.Actor.Address address, long uid) { } public Akka.Actor.Address Address { get; } - public int Uid { get; } + public long Uid { get; } public override Akka.Event.LogLevel LogLevel() { } public override string ToString() { } } @@ -219,7 +219,7 @@ namespace Akka.Remote public virtual void Init(Akka.Actor.Internal.ActorSystemImpl system) { } public Akka.Actor.IActorRef InternalResolveActorRef(string path) { } public Akka.Actor.Deploy LookUpRemotes(System.Collections.Generic.IEnumerable p) { } - public void Quarantine(Akka.Actor.Address address, int? uid) { } + public void Quarantine(Akka.Actor.Address address, long? uid) { } public void RegisterTempActor(Akka.Actor.IInternalActorRef actorRef, Akka.Actor.ActorPath path) { } public Akka.Actor.IActorRef ResolveActorRef(Akka.Actor.ActorPath actorPath) { } public Akka.Actor.IActorRef ResolveActorRef(string path) { } @@ -286,7 +286,7 @@ namespace Akka.Remote public abstract Akka.Actor.Address LocalAddressForRemote(Akka.Actor.Address remote); public abstract System.Threading.Tasks.Task ManagementCommand(object cmd); public abstract System.Threading.Tasks.Task ManagementCommand(object cmd, System.Threading.CancellationToken cancellationToken); - public abstract void Quarantine(Akka.Actor.Address address, int? uid); + public abstract void Quarantine(Akka.Actor.Address address, long? uid); public abstract void Send(object message, Akka.Actor.IActorRef sender, Akka.Remote.RemoteActorRef recipient); public abstract System.Threading.Tasks.Task Shutdown(); public abstract void Start(); @@ -308,7 +308,7 @@ namespace Akka.Remote protected override void OnReceive(object message) { } protected override void PostStop() { } protected virtual void PublishAddressTerminated(Akka.Actor.Address address) { } - protected virtual void Quarantine(Akka.Actor.Address address, int? addressUid) { } + protected virtual void Quarantine(Akka.Actor.Address address, long? addressUid) { } protected void RemoveWatch(Akka.Actor.IInternalActorRef watchee, Akka.Actor.IInternalActorRef watcher) { } protected void RemoveWatchee(Akka.Actor.IInternalActorRef watchee) { } protected void UnwatchNode(Akka.Actor.Address watcheeAddress) { } @@ -325,8 +325,8 @@ namespace Akka.Remote } public class HeartbeatRsp { - public HeartbeatRsp(int addressUid) { } - public int AddressUid { get; } + public HeartbeatRsp(long addressUid) { } + public long AddressUid { get; } } public class HeartbeatTick { From c7a2cd27204d61d91b98325a02d21876627dc31d Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 3 Jul 2026 18:23:52 -0500 Subject: [PATCH 5/9] widen-system-uid task group 8: compatibility tests + breaking-changes ledger entry - UniqueAddressWireCompatSpec: v1.5<->v1.6 gossip wire simulation - v1.6-emitted uint64 uid field read as uint32 (no truncation for uids <= uint32.MaxValue) and legacy uint32 bytes parsed by the widened v1.6 parser, plus serializer-level round-trips incl. >32-bit uids - AddressUidExtensionSpecs: default generation stays in [0, int.MaxValue]; use-64bit-system-uids = on yields nonzero uids; 64-draw statistical check that full-range generation escapes the legacy int range - >32-bit uid round-trips: MiscMessageSerializer (HeartbeatRsp), AkkaPduCodec (Associate handshake PDU), DData SerializationSupport (UniqueAddress) - BREAKING_CHANGES_V1.6.md: API + Wire rows for the int->long hard break (status Planned; flip to Merged with the PR link) --- BREAKING_CHANGES_V1.6.md | 2 + .../widen-system-uid-to-64bit/tasks.md | 6 +- .../UniqueAddressSerializationSupportSpec.cs | 57 ++++++++ .../UniqueAddressWireCompatSpec.cs | 136 ++++++++++++++++++ .../AddressUidExtensionSpecs.cs | 55 +++++++ .../MiscMessageSerializerSpec.cs | 13 ++ .../AkkaPduCodecFastDecodeDifferentialSpec.cs | 22 +++ 7 files changed, 288 insertions(+), 3 deletions(-) create mode 100644 src/contrib/cluster/Akka.DistributedData.Tests/Serialization/UniqueAddressSerializationSupportSpec.cs create mode 100644 src/core/Akka.Cluster.Tests/Serialization/UniqueAddressWireCompatSpec.cs diff --git a/BREAKING_CHANGES_V1.6.md b/BREAKING_CHANGES_V1.6.md index 8278aef9753..824218b83d5 100644 --- a/BREAKING_CHANGES_V1.6.md +++ b/BREAKING_CHANGES_V1.6.md @@ -40,4 +40,6 @@ of `Behavior`, `Wire`, `API` (combine with `+`). | Status | PR / Branch | Component | Type | Change | Migration | |--------|-------------|-----------|------|--------|-----------| +| Planned | `feature/widen-system-uid-to-64bit` | `Akka.Remote` / `Akka.Cluster` | API | System/address UID widened `int` → `long` across `AddressUid.Uid`, `AddressUidExtension.Uid()`, `UniqueAddress(Address, int)`/`.Uid`, `QuarantinedEvent(Address, int)`/`.Uid`, `RemoteWatcher.HeartbeatRsp(int)`/`.AddressUid`, and all `Quarantine(Address, int? uid)` members (`IRemoteActorRefProvider` / `RemoteTransport` / `RemoteWatcher`). Hard re-type (v1.6). Prerequisite for Artery. | Recompile against the `long` members; replace `int` uid locals with `long`. Default UID generation stays in 32-bit range; full 64-bit generation is opt-in via `akka.remote.use-64bit-system-uids = on`. | +| Planned | `feature/widen-system-uid-to-64bit` | `Akka.Cluster` | Wire | Cluster gossip `UniqueAddress.uid` widened `uint32` → `uint64` (`ClusterMessages.proto`). Same varint wire type → binary-compatible for uids ≤ `uint32.MaxValue`. | Rolling upgrade safe provided uid generation stays in 32-bit range (the default); enable `akka.remote.use-64bit-system-uids` only after the whole cluster + all remote peers are on v1.6. | | Planned | `feature/deprecate-actorpublisher` | `Akka.Streams` | Behavior | `Source.ActorRef` is re-implemented as a stream-native `GraphStage` (off the legacy `ActorPublisher`). Its public signature is unchanged, but the materialized `IActorRef` no longer treats `PoisonPill` / `Kill` as stream completion — those messages are now ignored. (`Status.Success` draining and `Status.Failure` completion are unchanged.) | Complete the stream explicitly: send `new Status.Success(...)` to drain buffered elements and complete, or `new Status.Failure(ex)` to fail. Do not rely on `PoisonPill` / `Kill`. | diff --git a/openspec/changes/widen-system-uid-to-64bit/tasks.md b/openspec/changes/widen-system-uid-to-64bit/tasks.md index e3f1a90a508..eea051c6e80 100644 --- a/openspec/changes/widen-system-uid-to-64bit/tasks.md +++ b/openspec/changes/widen-system-uid-to-64bit/tasks.md @@ -42,6 +42,6 @@ ## 8. Compatibility + docs -- [ ] 8.1 Rolling-upgrade test: v1.5 ↔ v1.6 gossip with int-range uids — no truncation, cluster forms -- [ ] 8.2 Full-range generation test (all-v1.6) — >32-bit uids round-trip -- [ ] 8.3 Add `BREAKING_CHANGES_V1.6.md` entry (API + Wire rows) +- [x] 8.1 Rolling-upgrade test: v1.5 ↔ v1.6 gossip with int-range uids — no truncation, cluster forms (wire-level simulation via uint32/uint64 varint cross-parse in `UniqueAddressWireCompatSpec`; a live mixed-version cluster cannot be exercised in-repo) +- [x] 8.2 Full-range generation test (all-v1.6) — >32-bit uids round-trip (gossip, heartbeat, handshake PDU, DData; generation range tests in `AddressUidExtensionSpecs`) +- [x] 8.3 Add `BREAKING_CHANGES_V1.6.md` entry (API + Wire rows) diff --git a/src/contrib/cluster/Akka.DistributedData.Tests/Serialization/UniqueAddressSerializationSupportSpec.cs b/src/contrib/cluster/Akka.DistributedData.Tests/Serialization/UniqueAddressSerializationSupportSpec.cs new file mode 100644 index 00000000000..79fc2b24d28 --- /dev/null +++ b/src/contrib/cluster/Akka.DistributedData.Tests/Serialization/UniqueAddressSerializationSupportSpec.cs @@ -0,0 +1,57 @@ +//----------------------------------------------------------------------- +// +// Copyright (C) 2009-2022 Lightbend Inc. +// Copyright (C) 2013-2025 .NET Foundation +// +//----------------------------------------------------------------------- + +using Akka.Actor; +using Akka.Configuration; +using Akka.DistributedData.Serialization; +using FluentAssertions; +using Xunit; +using Address = Akka.Actor.Address; +using UniqueAddress = Akka.Cluster.UniqueAddress; + +namespace Akka.DistributedData.Tests.Serialization +{ + /// + /// Task 8.2: DData's wire field was already int64 + /// (ReplicatorMessages.proto) before the widen-system-uid-to-64bit change - only the C# narrowing cast in + /// was removed. This verifies >32-bit uids + /// round-trip cleanly through / + /// . + /// + [Collection("DistributedDataSpec")] + public class UniqueAddressSerializationSupportSpec : TestKit.Xunit.TestKit + { + private static readonly Config BaseConfig = ConfigurationFactory.ParseString(@" + akka.actor { + provider=""Akka.Cluster.ClusterActorRefProvider, Akka.Cluster"" + } + akka.remote.dot-netty.tcp.port = 0").WithFallback(DistributedData.DefaultConfig()); + + private readonly SerializationSupport _support; + + public UniqueAddressSerializationSupportSpec(ITestOutputHelper output) + : base(BaseConfig, "UniqueAddressSerializationSupportSpec", output) + { + _support = new SerializationSupport((ExtendedActorSystem)Sys); + } + + [Theory(DisplayName = "Should_round_trip_unchanged_When_SerializationSupport_converts_a_64_bit_uid")] + [InlineData(12345L)] + [InlineData(long.MaxValue)] + [InlineData(unchecked((long)0x8000_0000_0000_0001))] // negative + [InlineData(1L << 40)] + public void Should_round_trip_unchanged_When_SerializationSupport_converts_a_64_bit_uid(long uid) + { + var address = new UniqueAddress(new Address("akka.tcp", Sys.Name, "some.host.org", 4711), uid); + + var proto = SerializationSupport.UniqueAddressToProto(address); + var roundTripped = _support.UniqueAddressFromProto(proto); + + roundTripped.Should().Be(address); + } + } +} diff --git a/src/core/Akka.Cluster.Tests/Serialization/UniqueAddressWireCompatSpec.cs b/src/core/Akka.Cluster.Tests/Serialization/UniqueAddressWireCompatSpec.cs new file mode 100644 index 00000000000..5eae1187d9c --- /dev/null +++ b/src/core/Akka.Cluster.Tests/Serialization/UniqueAddressWireCompatSpec.cs @@ -0,0 +1,136 @@ +//----------------------------------------------------------------------- +// +// Copyright (C) 2009-2022 Lightbend Inc. +// Copyright (C) 2013-2025 .NET Foundation +// +//----------------------------------------------------------------------- + +using System.IO; +using Akka.Actor; +using Akka.Cluster.Serialization; +using FluentAssertions; +using Google.Protobuf; +using Xunit; +using ProtoUniqueAddress = Akka.Cluster.Serialization.Proto.Msg.UniqueAddress; + +namespace Akka.Cluster.Tests.Serialization +{ + /// + /// Verifies the v1.5 <-> v1.6 rolling-upgrade wire compatibility of ClusterMessages.proto's + /// UniqueAddress.uid field, widened from uint32 to uint64 (same varint wire type, per + /// Decision 2/3 of the widen-system-uid-to-64bit design). A not-yet-upgraded v1.5 node must be able to read + /// a v1.6-emitted field (for uids <= uint32.MaxValue, i.e. default int-range generation), and a v1.6 node + /// must be able to read a legacy v1.5 uint32 field unchanged. + /// + /// These tests operate directly at the protobuf level with / + /// to simulate both sides of the gossip boundary without needing a live + /// v1.5 build. + /// + public class UniqueAddressWireCompatSpec + { + // Task 8.2: uids that only fit in the widened (>32-bit) range. + public static readonly long[] GreaterThan32BitUids = + { + long.MaxValue, + unchecked((long)0x8000_0000_0000_0001), // negative + 1L << 40, + }; + + [Theory(DisplayName = "Should_not_truncate_uid_When_a_v1_5_node_reads_a_v1_6_widened_UniqueAddress_field")] + [InlineData(0L)] + [InlineData(1L)] + [InlineData((long)int.MaxValue)] + [InlineData((long)uint.MaxValue)] + public void Should_not_truncate_uid_When_a_v1_5_node_reads_a_v1_6_widened_UniqueAddress_field(long uid) + { + // v1.6 builds + serializes the generated message, whose uid field is now a 64-bit (ulong) property. + var proto = new ProtoUniqueAddress { Uid = (ulong)uid }; + var bytes = proto.ToByteArray(); + + // A v1.5 node's generated parser reads field 2 as a uint32 - replicate that read by hand. + var parsedByV15 = ReadField2AsUInt32(bytes); + + parsedByV15.Should().Be((uint)uid); + } + + [Theory(DisplayName = "Should_read_legacy_uid_unchanged_When_a_v1_6_node_reads_a_v1_5_uint32_UniqueAddress_field")] + [InlineData(0L)] + [InlineData(1L)] + [InlineData((long)int.MaxValue)] + [InlineData((long)uint.MaxValue)] + public void Should_read_legacy_uid_unchanged_When_a_v1_6_node_reads_a_v1_5_uint32_UniqueAddress_field(long uid) + { + // Hand-write the wire bytes exactly as a v1.5 node's uint32 field would - proto3 fields are all + // optional, so writing ONLY field 2 (uid) is a valid, minimal UniqueAddress message on the wire. + var bytes = WriteField2AsUInt32((uint)uid); + + // v1.6's generated parser reads the same varint into its widened uint64 field. + var parsedByV16 = ProtoUniqueAddress.Parser.ParseFrom(bytes); + + parsedByV16.Uid.Should().Be((ulong)uid); + } + + [Fact(DisplayName = "Should_round_trip_unchanged_When_ClusterMessageSerializer_converts_an_int_range_uid")] + public void Should_round_trip_unchanged_When_ClusterMessageSerializer_converts_an_int_range_uid() + { + var address = new Address("akka.tcp", "system", "some.host.org", 4711); + var uniqueAddress = new UniqueAddress(address, 12345); + + var proto = ClusterMessageSerializer.UniqueAddressToProto(uniqueAddress); + var roundTripped = ClusterMessageSerializer.UniqueAddressFrom(proto); + + roundTripped.Should().Be(uniqueAddress); + } + + [Theory(DisplayName = "Should_round_trip_unchanged_When_ClusterMessageSerializer_converts_a_greater_than_32_bit_uid")] + [MemberData(nameof(GreaterThan32BitUidsMemberData))] + public void Should_round_trip_unchanged_When_ClusterMessageSerializer_converts_a_greater_than_32_bit_uid(long uid) + { + var address = new Address("akka.tcp", "system", "some.host.org", 4711); + var uniqueAddress = new UniqueAddress(address, uid); + + var proto = ClusterMessageSerializer.UniqueAddressToProto(uniqueAddress); + var roundTripped = ClusterMessageSerializer.UniqueAddressFrom(proto); + + roundTripped.Should().Be(uniqueAddress); + } + + public static TheoryData GreaterThan32BitUidsMemberData() + { + var data = new TheoryData(); + foreach (var uid in GreaterThan32BitUids) + data.Add(uid); + return data; + } + + private static uint ReadField2AsUInt32(byte[] bytes) + { + var cis = new CodedInputStream(bytes); + uint value = 0; + uint tag; + while ((tag = cis.ReadTag()) != 0) + { + if (WireFormat.GetTagFieldNumber(tag) == ProtoUniqueAddress.UidFieldNumber) + { + value = cis.ReadUInt32(); + } + else + { + cis.SkipLastField(); + } + } + + return value; + } + + private static byte[] WriteField2AsUInt32(uint value) + { + using var stream = new MemoryStream(); + var cos = new CodedOutputStream(stream); + cos.WriteTag(ProtoUniqueAddress.UidFieldNumber, WireFormat.WireType.Varint); + cos.WriteUInt32(value); + cos.Flush(); + return stream.ToArray(); + } + } +} diff --git a/src/core/Akka.Remote.Tests/AddressUidExtensionSpecs.cs b/src/core/Akka.Remote.Tests/AddressUidExtensionSpecs.cs index abe135eda19..a9fc73c5611 100644 --- a/src/core/Akka.Remote.Tests/AddressUidExtensionSpecs.cs +++ b/src/core/Akka.Remote.Tests/AddressUidExtensionSpecs.cs @@ -10,6 +10,7 @@ using System.Threading; using System.Threading.Tasks; using Akka.Actor; +using Akka.Configuration; using Akka.TestKit; using Xunit; @@ -48,5 +49,59 @@ public void AddressUidExtension_should_report_different_values_for_different_Act Shutdown(sys2); } } + + /// + /// Default uid generation (task 1.2) must stay in the legacy int range so a rolling upgrade from + /// v1.5 is safe (Decision 2 of the widen-system-uid-to-64bit design). + /// + [Fact] + public void AddressUidExtension_should_generate_a_uid_in_the_int_range_by_default() + { + var uid = AddressUidExtension.Uid(Sys); + Assert.True(uid >= 0 && uid <= int.MaxValue); + } + + /// + /// With akka.remote.use-64bit-system-uids = on, generation is opt-in full-range 64-bit + /// (task 1.3). A single draw could theoretically still land in the legacy int range, so this only + /// asserts the uid is nonzero (zero is reserved as a sentinel) - see + /// for the + /// statistical full-range check. + /// + [Fact] + public void AddressUidExtension_should_generate_a_nonzero_uid_when_64bit_uids_are_enabled() + { + var sys2 = ActorSystem.Create("AddressUidExtensionSpecs64Bit", + ConfigurationFactory.ParseString("akka.remote.use-64bit-system-uids = on")); + try + { + var uid = AddressUidExtension.Uid(sys2); + Assert.NotEqual(0, uid); + } + finally + { + Shutdown(sys2); + } + } + + /// + /// Statistical check that 's full-range 64-bit generator (task 1.2) actually + /// draws outside the legacy [0, int.MaxValue] range: each draw lands in that legacy range with + /// probability ~= 2^31 / 2^64 ~= 1.16e-10, so across 64 draws seeing zero out-of-range values is + /// effectively impossible unless the generator is broken (e.g. silently clamped back to int range). + /// + [Fact] + public void AddressUid_64bit_generation_should_draw_at_least_one_uid_outside_the_int_range() + { + const int draws = 64; + var uids = new long[draws]; + for (var i = 0; i < draws; i++) + { + uids[i] = new AddressUid(true).Uid; + } + + Assert.All(uids, uid => Assert.NotEqual(0, uid)); + Assert.Contains(uids, uid => uid < 0 || uid > int.MaxValue); + } } } diff --git a/src/core/Akka.Remote.Tests/Serialization/MiscMessageSerializerSpec.cs b/src/core/Akka.Remote.Tests/Serialization/MiscMessageSerializerSpec.cs index ae39aee28a9..f78f7148363 100644 --- a/src/core/Akka.Remote.Tests/Serialization/MiscMessageSerializerSpec.cs +++ b/src/core/Akka.Remote.Tests/Serialization/MiscMessageSerializerSpec.cs @@ -359,6 +359,19 @@ public void Can_serialize_RemoteWatcher_HearthbeatRsp() AssertAndReturn(heartbeatRsp).AddressUid.Should().Be(heartbeatRsp.AddressUid); //TODO: add Equals to RemoteWatcher.HeartbeatRsp } + // RemoteWatcher.HeartbeatRsp.AddressUid is carried on the wire as ContainerFormats.proto's uint64 + // (already 64-bit before the widen-system-uid-to-64bit change - only the C# narrowing casts were + // removed), so >32-bit uids must round-trip cleanly through MiscMessageSerializer. + [Theory(DisplayName = "Should_round_trip_a_greater_than_32_bit_addressUid_When_serializing_RemoteWatcher_HeartbeatRsp")] + [InlineData(long.MaxValue)] + [InlineData(unchecked((long)0x8000_0000_0000_0001))] // negative + [InlineData(1L << 40)] + public void Should_round_trip_a_greater_than_32_bit_addressUid_When_serializing_RemoteWatcher_HeartbeatRsp(long addressUid) + { + var heartbeatRsp = new RemoteWatcher.HeartbeatRsp(addressUid); + AssertAndReturn(heartbeatRsp).AddressUid.Should().Be(heartbeatRsp.AddressUid); + } + [Fact] public void Can_serialize_RemoteScope() { diff --git a/src/core/Akka.Remote.Tests/Transport/AkkaPduCodecFastDecodeDifferentialSpec.cs b/src/core/Akka.Remote.Tests/Transport/AkkaPduCodecFastDecodeDifferentialSpec.cs index 240fe50b8b6..cf21c2da55e 100644 --- a/src/core/Akka.Remote.Tests/Transport/AkkaPduCodecFastDecodeDifferentialSpec.cs +++ b/src/core/Akka.Remote.Tests/Transport/AkkaPduCodecFastDecodeDifferentialSpec.cs @@ -145,6 +145,28 @@ public void Decodes_ack_with_no_nacks() recipient: "alice", sender: "bob", seq: new SeqNo(1), payload: Payload(7, "m", 9), ack: Ack(3)); } + // ===================== handshake PDU: address/system UID round-trip ===================== + + [Theory(DisplayName = "Should_round_trip_a_64_bit_system_uid_When_encoding_and_decoding_an_Associate_PDU")] + [InlineData(12345L)] + [InlineData(long.MaxValue)] + [InlineData(unchecked((long)0x8000_0000_0000_0001))] // negative + [InlineData(1L << 40)] + public void Should_round_trip_a_64_bit_system_uid_When_encoding_and_decoding_an_Associate_PDU(long uid) + { + // The handshake's system/address uid is carried on the wire as WireFormats.proto's + // AkkaHandshakeInfo.uid (already fixed64 before the widen-system-uid-to-64bit change - only the + // C# narrowing casts were removed), so >32-bit uids must round-trip cleanly through + // ConstructAssociate -> DecodePdu. + var wire = _codec.ConstructAssociate(new HandshakeInfo(LocalAddress, uid)); + + var pdu = _codec.DecodePdu(wire); + + var associate = Assert.IsType(pdu); + Assert.Equal(LocalAddress, associate.Info.Origin); + Assert.Equal(uid, associate.Info.Uid); + } + // ===================== frozen interop wire bytes (golden) ===================== // The exact bytes classic Akka.Remote puts on the wire for the documented messages below. From 36b255ebb9e77a2d69bd288a4c249e5eea3b1453 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 3 Jul 2026 18:33:21 -0500 Subject: [PATCH 6/9] widen-system-uid task 7.3: Remote/Cluster/DData suites green (391/376/190, 0 failed) --- openspec/changes/widen-system-uid-to-64bit/tasks.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/openspec/changes/widen-system-uid-to-64bit/tasks.md b/openspec/changes/widen-system-uid-to-64bit/tasks.md index eea051c6e80..6198727cb89 100644 --- a/openspec/changes/widen-system-uid-to-64bit/tasks.md +++ b/openspec/changes/widen-system-uid-to-64bit/tasks.md @@ -38,7 +38,7 @@ - [x] 7.1 Update `Akka.API.Tests` approved files (`ApproveRemote.*`, `ApproveCluster.*`) for the hard break - [x] 7.2 `dotnet build -warnaserror` clean -- [ ] 7.3 Akka.Remote + Akka.Cluster + DistributedData tests green +- [x] 7.3 Akka.Remote + Akka.Cluster + DistributedData tests green (391 / 376 / 190 passed, 0 failed, net10.0 Release) ## 8. Compatibility + docs From e955723e2d9565571f92a33adc4fffb4d5e88290 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 3 Jul 2026 18:34:07 -0500 Subject: [PATCH 7/9] widen-system-uid: point breaking-changes ledger rows at PR #8317 --- BREAKING_CHANGES_V1.6.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/BREAKING_CHANGES_V1.6.md b/BREAKING_CHANGES_V1.6.md index 824218b83d5..7c09d3f0df0 100644 --- a/BREAKING_CHANGES_V1.6.md +++ b/BREAKING_CHANGES_V1.6.md @@ -40,6 +40,6 @@ of `Behavior`, `Wire`, `API` (combine with `+`). | Status | PR / Branch | Component | Type | Change | Migration | |--------|-------------|-----------|------|--------|-----------| -| Planned | `feature/widen-system-uid-to-64bit` | `Akka.Remote` / `Akka.Cluster` | API | System/address UID widened `int` → `long` across `AddressUid.Uid`, `AddressUidExtension.Uid()`, `UniqueAddress(Address, int)`/`.Uid`, `QuarantinedEvent(Address, int)`/`.Uid`, `RemoteWatcher.HeartbeatRsp(int)`/`.AddressUid`, and all `Quarantine(Address, int? uid)` members (`IRemoteActorRefProvider` / `RemoteTransport` / `RemoteWatcher`). Hard re-type (v1.6). Prerequisite for Artery. | Recompile against the `long` members; replace `int` uid locals with `long`. Default UID generation stays in 32-bit range; full 64-bit generation is opt-in via `akka.remote.use-64bit-system-uids = on`. | -| Planned | `feature/widen-system-uid-to-64bit` | `Akka.Cluster` | Wire | Cluster gossip `UniqueAddress.uid` widened `uint32` → `uint64` (`ClusterMessages.proto`). Same varint wire type → binary-compatible for uids ≤ `uint32.MaxValue`. | Rolling upgrade safe provided uid generation stays in 32-bit range (the default); enable `akka.remote.use-64bit-system-uids` only after the whole cluster + all remote peers are on v1.6. | +| Planned | [#8317](https://github.com/akkadotnet/akka.net/pull/8317) | `Akka.Remote` / `Akka.Cluster` | API | System/address UID widened `int` → `long` across `AddressUid.Uid`, `AddressUidExtension.Uid()`, `UniqueAddress(Address, int)`/`.Uid`, `QuarantinedEvent(Address, int)`/`.Uid`, `RemoteWatcher.HeartbeatRsp(int)`/`.AddressUid`, and all `Quarantine(Address, int? uid)` members (`IRemoteActorRefProvider` / `RemoteTransport` / `RemoteWatcher`). Hard re-type (v1.6). Prerequisite for Artery. | Recompile against the `long` members; replace `int` uid locals with `long`. Default UID generation stays in 32-bit range; full 64-bit generation is opt-in via `akka.remote.use-64bit-system-uids = on`. | +| Planned | [#8317](https://github.com/akkadotnet/akka.net/pull/8317) | `Akka.Cluster` | Wire | Cluster gossip `UniqueAddress.uid` widened `uint32` → `uint64` (`ClusterMessages.proto`). Same varint wire type → binary-compatible for uids ≤ `uint32.MaxValue`. | Rolling upgrade safe provided uid generation stays in 32-bit range (the default); enable `akka.remote.use-64bit-system-uids` only after the whole cluster + all remote peers are on v1.6. | | Planned | `feature/deprecate-actorpublisher` | `Akka.Streams` | Behavior | `Source.ActorRef` is re-implemented as a stream-native `GraphStage` (off the legacy `ActorPublisher`). Its public signature is unchanged, but the materialized `IActorRef` no longer treats `PoisonPill` / `Kill` as stream completion — those messages are now ignored. (`Status.Success` draining and `Status.Failure` completion are unchanged.) | Complete the stream explicitly: send `new Status.Success(...)` to drain buffered elements and complete, or `new Status.Failure(ex)` to fail. Do not rely on `PoisonPill` / `Kill`. | From 12b81463967b6a552e17c359a6552649afd393ec Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 3 Jul 2026 20:19:31 -0500 Subject: [PATCH 8/9] widen-system-uid: fix MNTR quarantine specs still expecting int uids CI (build 128643) failed 5 multi-node tests in Akka.Remote.Tests.MultiNode: the Subject actors reply with AddressUidExtension.Uid(...) - now long - but the driver-side helpers still asserted ExpectMsg/(int, IActorRef). Tell(object) erases the static type, so this compiled clean and only failed the runtime type check, on every run. Node2's port-bind failure in RemoteQuarantinePiercingSpec was collateral from the aborted barrier flow. Widened the ExpectMsg type params and uid locals to long in RemoteRestartedQuarantinedSpec, RemoteQuarantinePiercingSpec, and PiercingShouldKeepQuarantineSpec. No production change. All three specs pass locally through the MultiNode test adapter (6/6, net10.0 Release). --- .../PiercingShouldKeepQuarantineSpec.cs | 2 +- .../RemoteQuarantinePiercingSpec.cs | 8 ++++---- .../RemoteRestartedQuarantinedSpec.cs | 4 ++-- 3 files changed, 7 insertions(+), 7 deletions(-) diff --git a/src/core/Akka.Remote.Tests.MultiNode/PiercingShouldKeepQuarantineSpec.cs b/src/core/Akka.Remote.Tests.MultiNode/PiercingShouldKeepQuarantineSpec.cs index be80e81e749..c1256579fc8 100644 --- a/src/core/Akka.Remote.Tests.MultiNode/PiercingShouldKeepQuarantineSpec.cs +++ b/src/core/Akka.Remote.Tests.MultiNode/PiercingShouldKeepQuarantineSpec.cs @@ -78,7 +78,7 @@ public void While_probing_through_the_quarantine_remoting_must_not_lose_existing // Communicate with second system Sys.ActorSelection(Node(_config.Second) / "user" / "subject").Tell("getuid"); - var uid = ExpectMsg(TimeSpan.FromSeconds(10)); + var uid = ExpectMsg(TimeSpan.FromSeconds(10)); EnterBarrier("actor-identified"); // Manually Quarantine the other system diff --git a/src/core/Akka.Remote.Tests.MultiNode/RemoteQuarantinePiercingSpec.cs b/src/core/Akka.Remote.Tests.MultiNode/RemoteQuarantinePiercingSpec.cs index 1ab4be80584..f59c0615689 100644 --- a/src/core/Akka.Remote.Tests.MultiNode/RemoteQuarantinePiercingSpec.cs +++ b/src/core/Akka.Remote.Tests.MultiNode/RemoteQuarantinePiercingSpec.cs @@ -60,10 +60,10 @@ protected RemoteQuarantinePiercingSpec(RemoteQuarantinePiercingSpecConfig specCo _specConfig = specConfig; } - private (int, IActorRef) Identify(RoleName role, string actorName) + private (long, IActorRef) Identify(RoleName role, string actorName) { Sys.ActorSelection(Node(role) / "user" / actorName).Tell("identify"); - return ExpectMsg<(int, IActorRef)>(); + return ExpectMsg<(long, IActorRef)>(); } [MultiNodeFact] @@ -81,7 +81,7 @@ private void RemoteNodeShutdownAndComesBack_must_allow_piercing_through_the_quar // Acquire ActorRef from first system var tuple = Identify(_specConfig.Second, "subject"); - int uidFirst = tuple.Item1; + long uidFirst = tuple.Item1; IActorRef subjectFirst = tuple.Item2; EnterBarrier("actor-identified"); @@ -102,7 +102,7 @@ private void RemoteNodeShutdownAndComesBack_must_allow_piercing_through_the_quar AwaitAssert(() => { Sys.ActorSelection(new RootActorPath(secondAddress) / "user" / "subject").Tell("identify"); - var tuple2 = ExpectMsg<(int, IActorRef)>(TimeSpan.FromSeconds(1)); + var tuple2 = ExpectMsg<(long, IActorRef)>(TimeSpan.FromSeconds(1)); tuple2.Item1.Should().NotBe(uidFirst); tuple2.Item2.Should().NotBe(subjectFirst); }); diff --git a/src/core/Akka.Remote.Tests.MultiNode/RemoteRestartedQuarantinedSpec.cs b/src/core/Akka.Remote.Tests.MultiNode/RemoteRestartedQuarantinedSpec.cs index 6a496a36cff..73916b2c3d6 100644 --- a/src/core/Akka.Remote.Tests.MultiNode/RemoteRestartedQuarantinedSpec.cs +++ b/src/core/Akka.Remote.Tests.MultiNode/RemoteRestartedQuarantinedSpec.cs @@ -60,7 +60,7 @@ public Subject() public class RemoteRestartedQuarantinedSpec : MultiNodeSpec { private readonly RemoteRestartedQuarantinedMultiNetSpec _config; - private readonly Func _identifyWithUid; + private readonly Func _identifyWithUid; public RemoteRestartedQuarantinedSpec() : this(new RemoteRestartedQuarantinedMultiNetSpec()) @@ -75,7 +75,7 @@ protected RemoteRestartedQuarantinedSpec(RemoteRestartedQuarantinedMultiNetSpec _identifyWithUid = (role, actorName) => { Sys.ActorSelection(Node(role) / "user" / actorName).Tell("identify"); - return ExpectMsg<(int, IActorRef)>(); + return ExpectMsg<(long, IActorRef)>(); }; } From 3fa4f5307b336c955a97933296754e89aeeed45f Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 3 Jul 2026 20:36:43 -0500 Subject: [PATCH 9/9] widen-system-uid: convert quarantine MNTR specs to async TestKit methods Follow-up to 12b814639: the three quarantine specs still used blocking TestKit/MNTR calls. Converted per repo policy (no sync TestKit methods, no .Wait()/Thread.Sleep in tests): - ExpectMsg/ExpectNoMsg -> ExpectMsgAsync/ExpectNoMsgAsync - EnterBarrier -> EnterBarrierAsync; RunOn -> RunOnAsync - Within/AwaitAssert -> WithinAsync/AwaitAssertAsync - TestConductor.Shutdown(...).Wait(...) -> await TestConductor.ShutdownAsync(...) - WhenTerminated.Wait(...) -> await WhenTerminated.WaitAsync(...) - Thread.Sleep -> await Task.Delay - _identifyWithUid Func now Task-returning with async lambda No changes to test logic, assertions, barrier names, or timeouts. Clean -warnaserror rebuild; all three specs pass locally via the MultiNode adapter (6/6, net10.0 Release). --- .../PiercingShouldKeepQuarantineSpec.cs | 29 +++++++------- .../RemoteQuarantinePiercingSpec.cs | 39 ++++++++++--------- .../RemoteRestartedQuarantinedSpec.cs | 10 ++--- 3 files changed, 39 insertions(+), 39 deletions(-) diff --git a/src/core/Akka.Remote.Tests.MultiNode/PiercingShouldKeepQuarantineSpec.cs b/src/core/Akka.Remote.Tests.MultiNode/PiercingShouldKeepQuarantineSpec.cs index c1256579fc8..1b358a499dc 100644 --- a/src/core/Akka.Remote.Tests.MultiNode/PiercingShouldKeepQuarantineSpec.cs +++ b/src/core/Akka.Remote.Tests.MultiNode/PiercingShouldKeepQuarantineSpec.cs @@ -9,7 +9,6 @@ using System.Collections.Generic; using System.Linq; using System.Text; -using System.Threading; using System.Threading.Tasks; using Akka.Actor; using Akka.Configuration; @@ -65,46 +64,46 @@ protected override void OnReceive(object message) } [MultiNodeFact] - public void PiercingShouldKeepQuarantineSpecs() + public async Task PiercingShouldKeepQuarantineSpecs() { - While_probing_through_the_quarantine_remoting_must_not_lose_existing_quarantine_marker(); + await While_probing_through_the_quarantine_remoting_must_not_lose_existing_quarantine_marker(); } - public void While_probing_through_the_quarantine_remoting_must_not_lose_existing_quarantine_marker() + public async Task While_probing_through_the_quarantine_remoting_must_not_lose_existing_quarantine_marker() { - RunOn(() => + await RunOnAsync(async () => { - EnterBarrier("actors-started"); + await EnterBarrierAsync("actors-started"); // Communicate with second system Sys.ActorSelection(Node(_config.Second) / "user" / "subject").Tell("getuid"); - var uid = ExpectMsg(TimeSpan.FromSeconds(10)); - EnterBarrier("actor-identified"); + var uid = await ExpectMsgAsync(TimeSpan.FromSeconds(10)); + await EnterBarrierAsync("actor-identified"); // Manually Quarantine the other system RARP.For(Sys).Provider.Transport.Quarantine(Node(_config.Second).Address, uid); // Quarantining is not immediate - Thread.Sleep(1000); + await Task.Delay(1000); // Quarantine is up - Should not be able to communicate with remote system any more for (var i = 1; i <= 4; i++) { Sys.ActorSelection(Node(_config.Second) / "user" / "subject").Tell("getuid"); - ExpectNoMsg(TimeSpan.FromSeconds(2)); + await ExpectNoMsgAsync(TimeSpan.FromSeconds(2)); } - EnterBarrier("quarantine-intact"); + await EnterBarrierAsync("quarantine-intact"); }, _config.First); - RunOn(() => + await RunOnAsync(async () => { Sys.ActorOf("subject"); - EnterBarrier("actors-started"); - EnterBarrier("actor-identified"); - EnterBarrier("quarantine-intact"); + await EnterBarrierAsync("actors-started"); + await EnterBarrierAsync("actor-identified"); + await EnterBarrierAsync("quarantine-intact"); }, _config.Second); } } diff --git a/src/core/Akka.Remote.Tests.MultiNode/RemoteQuarantinePiercingSpec.cs b/src/core/Akka.Remote.Tests.MultiNode/RemoteQuarantinePiercingSpec.cs index f59c0615689..30421de7b0b 100644 --- a/src/core/Akka.Remote.Tests.MultiNode/RemoteQuarantinePiercingSpec.cs +++ b/src/core/Akka.Remote.Tests.MultiNode/RemoteQuarantinePiercingSpec.cs @@ -7,6 +7,7 @@ using System; +using System.Threading.Tasks; using Akka.Actor; using Akka.Configuration; using Akka.MultiNode.TestAdapter; @@ -60,49 +61,49 @@ protected RemoteQuarantinePiercingSpec(RemoteQuarantinePiercingSpecConfig specCo _specConfig = specConfig; } - private (long, IActorRef) Identify(RoleName role, string actorName) + private async Task<(long, IActorRef)> Identify(RoleName role, string actorName) { Sys.ActorSelection(Node(role) / "user" / actorName).Tell("identify"); - return ExpectMsg<(long, IActorRef)>(); + return await ExpectMsgAsync<(long, IActorRef)>(); } [MultiNodeFact] - public void RemoteQuarantinePiercingSpecs() + public async Task RemoteQuarantinePiercingSpecs() { - RemoteNodeShutdownAndComesBack_must_allow_piercing_through_the_quarantine_when_remote_UID_is_new(); + await RemoteNodeShutdownAndComesBack_must_allow_piercing_through_the_quarantine_when_remote_UID_is_new(); } - private void RemoteNodeShutdownAndComesBack_must_allow_piercing_through_the_quarantine_when_remote_UID_is_new() + private async Task RemoteNodeShutdownAndComesBack_must_allow_piercing_through_the_quarantine_when_remote_UID_is_new() { - RunOn(() => + await RunOnAsync(async () => { var secondAddress = Node(_specConfig.Second).Address; - EnterBarrier("actors-started"); + await EnterBarrierAsync("actors-started"); // Acquire ActorRef from first system - var tuple = Identify(_specConfig.Second, "subject"); + var tuple = await Identify(_specConfig.Second, "subject"); long uidFirst = tuple.Item1; IActorRef subjectFirst = tuple.Item2; - EnterBarrier("actor-identified"); + await EnterBarrierAsync("actor-identified"); // Manually Quarantine the other system RARP.For(Sys).Provider.Transport.Quarantine(Node(_specConfig.Second).Address, uidFirst); // Quarantine is up -- Cannot communicate with remote system any more Sys.ActorSelection(new RootActorPath(secondAddress) / "user" / "subject").Tell("identify"); - ExpectNoMsg(TimeSpan.FromSeconds(2)); + await ExpectNoMsgAsync(TimeSpan.FromSeconds(2)); // Shut down the other system -- which results in restart (see runOn(second)) - TestConductor.Shutdown(_specConfig.Second).Wait(TimeSpan.FromSeconds(30)); + await TestConductor.ShutdownAsync(_specConfig.Second); // Now wait until second system becomes alive again - Within(TimeSpan.FromSeconds(30), () => + await WithinAsync(TimeSpan.FromSeconds(30), async () => { // retry because the Subject actor might not be started yet - AwaitAssert(() => + await AwaitAssertAsync(async () => { Sys.ActorSelection(new RootActorPath(secondAddress) / "user" / "subject").Tell("identify"); - var tuple2 = ExpectMsg<(long, IActorRef)>(TimeSpan.FromSeconds(1)); + var tuple2 = await ExpectMsgAsync<(long, IActorRef)>(TimeSpan.FromSeconds(1)); tuple2.Item1.Should().NotBe(uidFirst); tuple2.Item2.Should().NotBe(subjectFirst); }); @@ -112,14 +113,14 @@ private void RemoteNodeShutdownAndComesBack_must_allow_piercing_through_the_quar Sys.ActorSelection(new RootActorPath(secondAddress) / "user" / "subject").Tell("shutdown"); }, _specConfig.First); - RunOn(() => + await RunOnAsync(async () => { var addr = ((ExtendedActorSystem)Sys).Provider.DefaultAddress; Sys.ActorOf(Props.Create(), "subject"); - EnterBarrier("actors-started"); + await EnterBarrierAsync("actors-started"); - EnterBarrier("actor-identified"); - Sys.WhenTerminated.Wait(TimeSpan.FromSeconds(30)); + await EnterBarrierAsync("actor-identified"); + await Sys.WhenTerminated.WaitAsync(TimeSpan.FromSeconds(30)); var freshSystem = ActorSystem.Create(Sys.Name, ConfigurationFactory.ParseString($@" akka.remote.dot-netty.tcp.hostname = {addr.Host} @@ -127,7 +128,7 @@ private void RemoteNodeShutdownAndComesBack_must_allow_piercing_through_the_quar ").WithFallback(Sys.Settings.Config)); freshSystem.ActorOf(Props.Create(), "subject"); - freshSystem.WhenTerminated.Wait(TimeSpan.FromSeconds(30)); + await freshSystem.WhenTerminated.WaitAsync(TimeSpan.FromSeconds(30)); }, _specConfig.Second); } } diff --git a/src/core/Akka.Remote.Tests.MultiNode/RemoteRestartedQuarantinedSpec.cs b/src/core/Akka.Remote.Tests.MultiNode/RemoteRestartedQuarantinedSpec.cs index 73916b2c3d6..d4366c7a644 100644 --- a/src/core/Akka.Remote.Tests.MultiNode/RemoteRestartedQuarantinedSpec.cs +++ b/src/core/Akka.Remote.Tests.MultiNode/RemoteRestartedQuarantinedSpec.cs @@ -60,7 +60,7 @@ public Subject() public class RemoteRestartedQuarantinedSpec : MultiNodeSpec { private readonly RemoteRestartedQuarantinedMultiNetSpec _config; - private readonly Func _identifyWithUid; + private readonly Func> _identifyWithUid; public RemoteRestartedQuarantinedSpec() : this(new RemoteRestartedQuarantinedMultiNetSpec()) @@ -72,10 +72,10 @@ protected RemoteRestartedQuarantinedSpec(RemoteRestartedQuarantinedMultiNetSpec { _config = config; - _identifyWithUid = (role, actorName) => + _identifyWithUid = async (role, actorName) => { Sys.ActorSelection(Node(role) / "user" / actorName).Tell("identify"); - return ExpectMsg<(long, IActorRef)>(); + return await ExpectMsgAsync<(long, IActorRef)>(); }; } @@ -90,7 +90,7 @@ public async Task A_restarted_quarantined_system_should_not_crash_the_other_syst await RunOnAsync(async () => { var secondAddress = Node(_config.Second).Address; - var uid = _identifyWithUid(_config.Second, "subject").Item1; + var uid = (await _identifyWithUid(_config.Second, "subject")).Item1; RARP.For(Sys).Provider.Transport.Quarantine(Node(_config.Second).Address, uid); @@ -118,7 +118,7 @@ await RunOnAsync(async () => var firstAddress = Node(_config.First).Address; Sys.EventStream.Subscribe(TestActor, typeof (ThisActorSystemQuarantinedEvent)); - var actorRef = _identifyWithUid(_config.First, "subject").Item2; + var actorRef = (await _identifyWithUid(_config.First, "subject")).Item2; await EnterBarrierAsync("quarantined");