diff --git a/BREAKING_CHANGES_V1.6.md b/BREAKING_CHANGES_V1.6.md index 8278aef9753..7c09d3f0df0 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 | [#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`. | 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. diff --git a/openspec/changes/widen-system-uid-to-64bit/tasks.md b/openspec/changes/widen-system-uid-to-64bit/tasks.md index b30759afeae..6198727cb89 100644 --- a/openspec/changes/widen-system-uid-to-64bit/tasks.md +++ b/openspec/changes/widen-system-uid-to-64bit/tasks.md @@ -1,47 +1,47 @@ ## 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 -- [ ] 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`) -- [ ] 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.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 -- [ ] 7.1 Update `Akka.API.Tests` approved files (`ApproveRemote.*`, `ApproveCluster.*`) for the hard break -- [ ] 7.2 `dotnet build -warnaserror` clean -- [ ] 7.3 Akka.Remote + Akka.Cluster + DistributedData tests green +- [x] 7.1 Update `Akka.API.Tests` approved files (`ApproveRemote.*`, `ApproveCluster.*`) for the hard break +- [x] 7.2 `dotnet build -warnaserror` clean +- [x] 7.3 Akka.Remote + Akka.Cluster + DistributedData tests green (391 / 376 / 190 passed, 0 failed, net10.0 Release) ## 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/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.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 { 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.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/core/Akka.Remote.Tests.MultiNode/PiercingShouldKeepQuarantineSpec.cs b/src/core/Akka.Remote.Tests.MultiNode/PiercingShouldKeepQuarantineSpec.cs index be80e81e749..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 1ab4be80584..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 (int, 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<(int, 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"); - int uidFirst = tuple.Item1; + 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<(int, 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 6a496a36cff..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<(int, 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"); diff --git a/src/core/Akka.Remote.Tests/AddressUidExtensionSpecs.cs b/src/core/Akka.Remote.Tests/AddressUidExtensionSpecs.cs index 7fabdf2cb81..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; @@ -23,7 +24,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, _ => { @@ -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/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.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. 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)); } 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; } /****************************************