diff --git a/CLAUDE.md b/CLAUDE.md index 07780f9..1b2ab05 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -152,24 +152,41 @@ Part: ### Handshake (the easiest place to get burned) The handshake does **not** use the standard frame format — it's an ad-hoc -byte sequence. Translate it byte-for-byte from -`cppcache/src/TcrConnection.cpp::sendHandshakeForServer`. **Do not work -from memory.** +byte sequence. **Authoritative reference is the Java code, not cppcache** — +when they disagree, the Java server wins: + +- client side: `geode-core/.../cache/client/internal/ClientSideHandshakeImpl.java::write` +- server side: `geode-core/.../cache/tier/sockets/ServerSideHandshakeImpl.java` +- shared : `geode-core/.../cache/tier/sockets/Handshake.java` (constants, helpers) + +cppcache `TcrConnection.cpp::sendHandshakeForServer` is a parallel +implementation with stale comments; cross-check before trusting it. **Do +not work from memory.** ``` client → server: - ConnectionType u8 (100 = client-to-server) - ReplyOk u8 (59) - ProtocolVersion (major.minor.patch + ordinal) - ClientProxyMembershipID (serialised: host / PID / UUID / durable id) - Credentials (optional Properties) + ConnectionType u8 (100 = CLIENT_TO_SERVER, 101/102 = notification) + ProtocolVersion (ordinal only; 1 byte if ≤ 127, else sentinel + i16) + ReplyOk u8 (59) + ReadTimeout i32 (request/response only; notification writes port list instead) + ClientProxyMembershipID (one DataSerializable object on the wire: + FixedIDByte u8 = 1 + DSFid u8 = 38 + identity varint length + bytes + uniqueId i32) + Overrides[] u8 × N (currently always N = 1: conflation byte) + SecurityMode u8 (0 = none, 1 = normal + creds body, 3 = multi-user notification) + [Credentials body] (only when SecurityMode != none) server → client: - AcceptanceCode u8 (38 = OK) - ServerQueueStatus u8 - QueueSize i32 - ServerMember (membership ID) - DeltaEnabled u8 + AcceptanceCode u8 (59 = OK; 60 REFUSED / 61 INVALID / 66 AUTH_NOT_REQUIRED / + 67 SERVER_IS_LOCATOR / 21 SSL_REQUIRED on rejection) + EndpointType u8 (subscription/queue role — drain in MVP) + QueueSize i32 (subscription queue size — drain in MVP) + ServerMember (DataSerializable membership ID — drain in MVP) + Message (UTF-8 str) (server diagnostic / refusal text; empty on success, + u16 length prefix) + DeltaEnabled u8 (bool) (delta propagation flag — drain in MVP) ``` ### MVP MessageType subset diff --git a/Directory.Packages.props b/Directory.Packages.props index 99235b9..199afd9 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -20,6 +20,7 @@ + @@ -27,4 +28,4 @@ - \ No newline at end of file + diff --git a/src/Geode.Client/Geode.Client.csproj b/src/Geode.Client/Geode.Client.csproj index 2c81404..437e843 100644 --- a/src/Geode.Client/Geode.Client.csproj +++ b/src/Geode.Client/Geode.Client.csproj @@ -13,6 +13,10 @@ alpha.0 + + 1701;1702;CA1873 + + diff --git a/src/Geode.Client/GeodeClientServiceCollectionExtensions.cs b/src/Geode.Client/GeodeClientServiceCollectionExtensions.cs new file mode 100644 index 0000000..51e2596 --- /dev/null +++ b/src/Geode.Client/GeodeClientServiceCollectionExtensions.cs @@ -0,0 +1,63 @@ +using Geode.Client.Options; +using Geode.Client.Protocol; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; + +namespace Geode.Client; + +/// +/// DI registration entry point for the Geode managed client. +/// +public static class GeodeClientServiceCollectionExtensions +{ + /// + /// Register the Geode client services and bind + /// from + /// (typically the "Geode" + /// section of appsettings.json). + /// + /// + /// + /// builder.Services.AddGeodeClient( + /// builder.Configuration.GetSection("Geode")); + /// + /// + /// + /// + /// Phase 2 surface — registers the bare minimum needed to open and + /// handshake a single connection: + /// + /// + /// bound from configuration. + /// + /// as a singleton + /// — process-scoped uniqueTag and identity-bytes cache must be + /// shared across all connections. + /// + /// + /// as transient — every borrow + /// yields a fresh connection. Phase 6 will replace this with a + /// pooled lifetime. + /// + /// + /// + /// Logging is intentionally not registered here; callers are expected + /// to add their own ILoggerFactory via AddLogging() / + /// AddHttpLogging() / Serilog / etc. so the Geode client picks + /// up whatever logging stack the host already configured. + /// + /// + public static IServiceCollection AddGeodeClient( + this IServiceCollection services, + IConfiguration configuration) + { + ArgumentNullException.ThrowIfNull(services); + ArgumentNullException.ThrowIfNull(configuration); + + services.AddOptions().Bind(configuration); + services.AddSingleton(); + services.AddTransient(); + + return services; + } +} diff --git a/src/Geode.Client/GeodeException.cs b/src/Geode.Client/GeodeException.cs new file mode 100644 index 0000000..7d54601 --- /dev/null +++ b/src/Geode.Client/GeodeException.cs @@ -0,0 +1,30 @@ +namespace Geode.Client; + +/// +/// Base exception for Geode-specific protocol-level failures: server-side +/// refusals (e.g. handshake rejection), malformed wire bytes, and exceptions +/// returned by the server in MessageType.Exception replies. +/// +/// +/// +/// Distinct from / +/// which surface for +/// genuine transport failures, and from +/// which is reserved for API +/// misuse (e.g. SendAsync before ConnectAsync). +/// +/// +/// Catch this type to handle "the Geode server said something we couldn't +/// proceed with" without swallowing unrelated BCL failures. +/// +/// +public class GeodeException : Exception +{ + public GeodeException() { } + + public GeodeException(string message) + : base(message) { } + + public GeodeException(string message, Exception innerException) + : base(message, innerException) { } +} diff --git a/src/Geode.Client/Options/GeodeClientOptions.cs b/src/Geode.Client/Options/GeodeClientOptions.cs new file mode 100644 index 0000000..c4d7f68 --- /dev/null +++ b/src/Geode.Client/Options/GeodeClientOptions.cs @@ -0,0 +1,47 @@ +namespace Geode.Client.Options; + +/// +/// User-facing configuration for the Geode client. Bound from the +/// "Geode" section of appsettings.json via +/// IOptions<GeodeClientOptions> and consumed by the (Phase 5) +/// AddGeodeClient(...) DI extension. +/// +/// +/// +/// Property set is derived from cppcache SystemProperties (file +/// cppcache/include/geode/SystemProperties.hpp + defaults in +/// cppcache/src/SystemProperties.cpp). The following cppcache +/// fields are intentionally omitted because the .NET runtime / +/// our architecture replaces them: +/// +/// +/// statistic-* (use EventCounters / OpenTelemetry). +/// log-* (use ILogger + filter levels). +/// heap-lru-* / tombstone-timeout (server-side concepts). +/// suspended-tx-timeout / bucket-wait-timeout (out of MVP scope). +/// max-fe-threads / enable-chunk-handler-thread (.NET ThreadPool managed). +/// security-client-dhalgo (Diffie-Hellman creds — deprecated upstream). +/// on-client-disconnect-clear-pdxType-Ids (Phase 11 PDX). +/// cache-xml-file (CLAUDE.md cuts cache.xml entirely). +/// +/// +public class GeodeClientOptions +{ + /// + /// Distributed-system / client name shown in server logs. Mirrors + /// cppcache name. Default empty. + /// + public string Name { get; set; } = string.Empty; + + /// Connection-pool tuning. See . + public PoolOptions Pool { get; } = new(); + + /// TLS / SSL settings. See . + public TlsOptions Tls { get; } = new(); + + /// + /// Subscription / durable-client / event-notification settings. + /// See . + /// + public SubscriptionOptions Subscription { get; } = new(); +} diff --git a/src/Geode.Client/Options/PoolOptions.cs b/src/Geode.Client/Options/PoolOptions.cs new file mode 100644 index 0000000..b457e66 --- /dev/null +++ b/src/Geode.Client/Options/PoolOptions.cs @@ -0,0 +1,56 @@ +namespace Geode.Client.Options; + +/// +/// Connection-pool tuning derived from cppcache +/// SystemProperties. Defaults match cppcache's own constants in +/// SystemProperties.cpp so behaviour is interchangeable until we +/// have reason to diverge. +/// +public class PoolOptions +{ + /// + /// Number of TCP connections to maintain in the pool. Mirrors cppcache + /// connection-pool-size; default 5. + /// + /// + /// Phase 6 (pool) consumer. CLAUDE.md schema splits this into + /// MinConnections / MaxConnections; for now we expose a + /// single fixed size like cppcache and revisit when the pool is built. + /// + public int ConnectionPoolSize { get; set; } = 5; + + /// + /// Time budget for the TCP connect + handshake. Mirrors cppcache + /// connect-timeout; default 59 seconds. + /// + public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(59); + + /// + /// Extra wait between failed connect attempts. Mirrors cppcache + /// connect-wait-timeout; default + /// (= disabled). Linux-specific in cppcache; kept here for parity but + /// likely unused by .NET socket APIs. + /// + public TimeSpan ConnectWaitTimeout { get; set; } = TimeSpan.Zero; + + /// + /// Send / receive buffer size hint for the underlying socket. Mirrors + /// cppcache max-socket-buffer-size; default 65 × 1024 = 66560 bytes. + /// + public int MaxSocketBufferSize { get; set; } = 65 * 1024; + + /// + /// Idle keep-alive ping cadence. Mirrors cppcache ping-interval; + /// default 10 seconds. The pool sends a MessageType.Ping on idle + /// connections at this rate so the server doesn't time them out. + /// + public TimeSpan PingInterval { get; set; } = TimeSpan.FromSeconds(10); + + /// + /// Whether to randomise the order in which servers are tried. + /// cppcache uses the inverted disable-shuffling-of-endpoints + /// (default false ⇒ shuffle by default), so the equivalent default here + /// is true. + /// + public bool ShuffleEndpoints { get; set; } = true; +} diff --git a/src/Geode.Client/Options/SubscriptionOptions.cs b/src/Geode.Client/Options/SubscriptionOptions.cs new file mode 100644 index 0000000..6274725 --- /dev/null +++ b/src/Geode.Client/Options/SubscriptionOptions.cs @@ -0,0 +1,72 @@ +namespace Geode.Client.Options; + +/// +/// Subscription, durable-client, and event-notification settings. +/// Mirrors the subscription-related fields of cppcache +/// SystemProperties. The whole group is dormant until Phase 12+ +/// adds CQ / register-interest / event listeners. +/// +public class SubscriptionOptions +{ + /// + /// Stable client identifier that lets the server retain this client's + /// subscription queue across reconnects. Mirrors cppcache + /// durable-client-id; default empty (= non-durable, server + /// discards the queue on disconnect). + /// + /// + /// Set a stable string (e.g. "order-service-pod-1") to opt into + /// durable subscriptions. Consumed by the + /// ClientProxyMembershipID builder when subscriptions ship in + /// Phase 12+. + /// + public string DurableClientId { get; set; } = string.Empty; + + /// + /// How long the server should retain this client's subscription queue + /// after a disconnect before giving up. Mirrors cppcache + /// durable-timeout; default 300 seconds. Only meaningful when + /// is set. + /// + public TimeSpan DurableTimeout { get; set; } = TimeSpan.FromSeconds(300); + + /// + /// Whether a non-durable client starts receiving subscription events + /// automatically once regions are created. Mirrors cppcache + /// auto-ready-for-events; default true. Set to + /// false to require an explicit "ready" call after wiring up + /// listeners (Phase 12+ API). + /// + public bool AutoReadyForEvents { get; set; } = true; + + /// + /// How often the client checks subscription redundancy (HA queue copy + /// count). Mirrors cppcache redundancy-monitor-interval; + /// default 10 seconds. + /// + public TimeSpan RedundancyMonitorInterval { get; set; } = TimeSpan.FromSeconds(10); + + /// + /// Periodic ack cadence for received subscription notifications. + /// Mirrors cppcache notify-ack-interval; default 1 second. + /// + public TimeSpan NotifyAckInterval { get; set; } = TimeSpan.FromSeconds(1); + + /// + /// How long an idle event-id map entry is kept for duplicate-event + /// detection on the subscription channel. Mirrors cppcache + /// notify-dupcheck-life; default 300 seconds. + /// + public TimeSpan NotifyDupCheckLife { get; set; } = TimeSpan.FromSeconds(300); + + /// + /// Per-client event-conflation override sent in the handshake's + /// "overrides" byte. Tristate: null (default) defers to the + /// server-side setting, true forces conflation on for this + /// client, false forces it off. Mirrors cppcache + /// conflate-events's string values "server" / + /// "true" / "false", but bool? is the type-safe + /// way to express the same three states in C#. + /// + public bool? ConflateEvents { get; set; } +} diff --git a/src/Geode.Client/Options/TlsOptions.cs b/src/Geode.Client/Options/TlsOptions.cs new file mode 100644 index 0000000..e2f6185 --- /dev/null +++ b/src/Geode.Client/Options/TlsOptions.cs @@ -0,0 +1,34 @@ +namespace Geode.Client.Options; + +/// +/// TLS / SSL configuration. Mirrors cppcache ssl-* settings but +/// will eventually layer on top of System.Net.Security.SslStream +/// (Phase 8) — file paths may be replaced or augmented with +/// X509Certificate2 handles when we get there. +/// +public class TlsOptions +{ + /// + /// Whether to upgrade the socket with TLS after TCP connect. Mirrors + /// cppcache ssl-enabled; default false. + /// + public bool Enabled { get; set; } + + /// + /// Path to the client keystore (.pem in cppcache). Mirrors cppcache + /// ssl-keystore; default empty. + /// + public string KeyStorePath { get; set; } = string.Empty; + + /// + /// Password protecting the keystore at . + /// Mirrors cppcache ssl-keystore-password; default empty. + /// + public string KeyStorePassword { get; set; } = string.Empty; + + /// + /// Path to the truststore used to validate the server certificate + /// chain. Mirrors cppcache ssl-truststore; default empty. + /// + public string TrustStorePath { get; set; } = string.Empty; +} diff --git a/src/Geode.Client/Protocol/BigEndianBinaryReader.cs b/src/Geode.Client/Protocol/BigEndianBinaryReader.cs index 047ad7b..9fd7474 100644 --- a/src/Geode.Client/Protocol/BigEndianBinaryReader.cs +++ b/src/Geode.Client/Protocol/BigEndianBinaryReader.cs @@ -92,32 +92,65 @@ public ReadOnlyMemory ReadBytesOnly(int count) // ====================================================================== /// Read a signed 8-bit integer (i8). - public sbyte ReadSByte() => - throw new NotImplementedException("Phase 2 handshake."); + /// + /// Two's-complement reinterpretation of the next wire byte (e.g. 0xFF + /// → -1), matching what Java's DataInput::readByte returns. + /// + public sbyte ReadSByte() => (sbyte)ReadByte(); /// Read a 16-bit signed integer in big-endian byte order. - public short ReadInt16() => - throw new NotImplementedException("Phase 2 handshake / Phase 4 typed values."); + public short ReadInt16() + { + EnsureAvailable(sizeof(short)); + var value = BinaryPrimitives.ReadInt16BigEndian(_buffer.Span.Slice(_position, sizeof(short))); + _position += sizeof(short); + return value; + } /// Read a 16-bit unsigned integer in big-endian byte order. Mirrors cppcache readChar. - public ushort ReadUInt16() => - throw new NotImplementedException("Phase 4 typed values."); + public ushort ReadUInt16() + { + EnsureAvailable(sizeof(ushort)); + var value = BinaryPrimitives.ReadUInt16BigEndian(_buffer.Span.Slice(_position, sizeof(ushort))); + _position += sizeof(ushort); + return value; + } /// Read a 32-bit unsigned integer in big-endian byte order. - public uint ReadUInt32() => - throw new NotImplementedException("Phase 4 typed values."); + public uint ReadUInt32() + { + EnsureAvailable(sizeof(uint)); + var value = BinaryPrimitives.ReadUInt32BigEndian(_buffer.Span.Slice(_position, sizeof(uint))); + _position += sizeof(uint); + return value; + } /// Read a 64-bit unsigned integer in big-endian byte order. - public ulong ReadUInt64() => - throw new NotImplementedException("Phase 4 typed values."); + public ulong ReadUInt64() + { + EnsureAvailable(sizeof(ulong)); + var value = BinaryPrimitives.ReadUInt64BigEndian(_buffer.Span.Slice(_position, sizeof(ulong))); + _position += sizeof(ulong); + return value; + } /// Read an IEEE 754 single-precision float in big-endian byte order. - public float ReadFloat() => - throw new NotImplementedException("Phase 4 typed values."); + public float ReadFloat() + { + EnsureAvailable(sizeof(float)); + var value = BinaryPrimitives.ReadSingleBigEndian(_buffer.Span.Slice(_position, sizeof(float))); + _position += sizeof(float); + return value; + } /// Read an IEEE 754 double-precision float in big-endian byte order. - public double ReadDouble() => - throw new NotImplementedException("Phase 4 typed values."); + public double ReadDouble() + { + EnsureAvailable(sizeof(double)); + var value = BinaryPrimitives.ReadDoubleBigEndian(_buffer.Span.Slice(_position, sizeof(double))); + _position += sizeof(double); + return value; + } /// /// Read a length-prefixed byte sequence: i32 length followed by the bytes. diff --git a/src/Geode.Client/Protocol/BigEndianBinaryWriter.cs b/src/Geode.Client/Protocol/BigEndianBinaryWriter.cs index 0835c47..b0cb86e 100644 --- a/src/Geode.Client/Protocol/BigEndianBinaryWriter.cs +++ b/src/Geode.Client/Protocol/BigEndianBinaryWriter.cs @@ -65,47 +65,112 @@ public void WriteInt64(long value) // ====================================================================== /// Write a signed 8-bit integer (i8). - public void WriteSByte(sbyte value) => - throw new NotImplementedException("Phase 2 handshake."); + /// + /// Two's-complement reinterpretation: (byte)value produces the same + /// bit pattern that Java's DataOutput::writeByte writes for an + /// int8_t (e.g. -1 → 0xFF). + /// + public void WriteSByte(sbyte value) => _buffer.WriteByte((byte)value); /// Write a 16-bit signed integer in big-endian byte order. - public void WriteInt16(short value) => - throw new NotImplementedException("Phase 2 handshake / Phase 4 typed values."); + public void WriteInt16(short value) + { + Span tmp = stackalloc byte[sizeof(short)]; + BinaryPrimitives.WriteInt16BigEndian(tmp, value); + _buffer.Write(tmp); + } /// Write a 16-bit unsigned integer in big-endian byte order. Mirrors cppcache writeChar. - public void WriteUInt16(ushort value) => - throw new NotImplementedException("Phase 4 typed values."); + public void WriteUInt16(ushort value) + { + Span tmp = stackalloc byte[sizeof(ushort)]; + BinaryPrimitives.WriteUInt16BigEndian(tmp, value); + _buffer.Write(tmp); + } /// Write a 32-bit unsigned integer in big-endian byte order. - public void WriteUInt32(uint value) => - throw new NotImplementedException("Phase 4 typed values."); + public void WriteUInt32(uint value) + { + Span tmp = stackalloc byte[sizeof(uint)]; + BinaryPrimitives.WriteUInt32BigEndian(tmp, value); + _buffer.Write(tmp); + } /// Write a 64-bit unsigned integer in big-endian byte order. - public void WriteUInt64(ulong value) => - throw new NotImplementedException("Phase 4 typed values."); + public void WriteUInt64(ulong value) + { + Span tmp = stackalloc byte[sizeof(ulong)]; + BinaryPrimitives.WriteUInt64BigEndian(tmp, value); + _buffer.Write(tmp); + } /// Write an IEEE 754 single-precision float in big-endian byte order. - public void WriteFloat(float value) => - throw new NotImplementedException("Phase 4 typed values."); + public void WriteFloat(float value) + { + Span tmp = stackalloc byte[sizeof(float)]; + BinaryPrimitives.WriteSingleBigEndian(tmp, value); + _buffer.Write(tmp); + } /// Write an IEEE 754 double-precision float in big-endian byte order. - public void WriteDouble(double value) => - throw new NotImplementedException("Phase 4 typed values."); + public void WriteDouble(double value) + { + Span tmp = stackalloc byte[sizeof(double)]; + BinaryPrimitives.WriteDoubleBigEndian(tmp, value); + _buffer.Write(tmp); + } /// - /// Write a length-prefixed byte sequence: i32 length followed by the bytes, - /// or i32 -1 if is null. + /// Write a length-prefixed byte sequence: + /// length (varint) followed by the bytes, or a single -1 sentinel + /// byte if is null. /// Mirrors cppcache DataOutput::writeBytes. /// - public void WriteBytes(byte[]? bytes) => - throw new NotImplementedException("Phase 3 Put/Get value parts."); + public void WriteBytes(byte[]? bytes) + { + if (bytes is null) + { + WriteArrayLen(-1); + return; + } + WriteArrayLen(bytes.Length); + _buffer.Write(bytes); + } /// - /// Write Geode's variable-length array length encoding (1, 2, or 4 bytes - /// depending on magnitude). Mirrors cppcache DataOutput::writeArrayLen. + /// Write Geode's variable-length array-length encoding (1, 3, or 5 bytes + /// total). Mirrors cppcache DataOutput::writeArrayLen. /// - public void WriteArrayLen(int length) => - throw new NotImplementedException("Phase 4 collection-bearing parts."); + /// + /// Encoding (matches Java collection-length convention): + /// + /// length == -1 → 1 byte: 0xFF (null sentinel). + /// length ≤ 252 → 1 byte: the length itself. + /// length ≤ 0xFFFF → 3 bytes: 0xFE + u16 length. + /// otherwise (up to int.MaxValue) → 5 bytes: 0xFD + i32 length. + /// + /// + public void WriteArrayLen(int length) + { + if (length == -1) + { + WriteSByte(-1); + } + else if (length <= 252) + { + WriteByte((byte)length); + } + else if (length <= 0xFFFF) + { + WriteSByte(-2); + WriteUInt16((ushort)length); + } + else + { + WriteSByte(-3); + WriteInt32(length); + } + } /// /// Write a string in Java modified UTF-8 with a u16 byte-length prefix. @@ -118,8 +183,135 @@ public void WriteArrayLen(int length) => /// surrogate written as a 3-byte sequence (so a single supplementary /// codepoint takes 6 bytes, not 4 as in standard UTF-8). /// - public void WriteJavaModifiedUtf8(string? value) => - throw new NotImplementedException("Phase 4 string values."); + /// + /// Write a Geode-tagged string: [DSCode byte][body]. Mirrors + /// cppcache DataOutput::writeString; the matching reader on the + /// server is StaticSerialization.readString, which switches on + /// the leading DSCode byte. + /// + /// + /// Branches: + /// + /// null → 1 byte: CacheableNullString (69). + /// + /// All ASCII (no NUL, all chars ≤ 0x7F), length ≤ 0xFFFF → + /// CacheableASCIIString (87) + u16 length + ASCII bytes. + /// + /// + /// Has non-ASCII chars, modified-UTF-8 byte length ≤ 0xFFFF → + /// CacheableString (42) + u16 byte-length + modified-UTF-8 bytes. + /// + /// + /// Lengths exceeding 0xFFFF map to the *Huge DSCode + /// variants (88 / 89). Not implemented yet — throws; fill in when a + /// wire field with a huge string actually appears. + /// + /// + /// + public void WriteString(string? value) + { + const byte CacheableString = 42; + const byte CacheableNullString = 69; + const byte CacheableAsciiString = 87; + + if (value is null) + { + WriteByte(CacheableNullString); + return; + } + + var hasNonAscii = false; + foreach (var c in value) + { + if (c == 0 || c > 0x007F) + { + hasNonAscii = true; + break; + } + } + + if (hasNonAscii) + { + // CacheableString: leading byte + u16 byte-length + modified UTF-8. + // WriteJavaModifiedUtf8 already emits the u16 prefix + body, so + // we just stamp the DSCode in front and delegate. + WriteByte(CacheableString); + WriteJavaModifiedUtf8(value); + return; + } + + if (value.Length > 0xFFFF) + { + throw new NotImplementedException( + $"CacheableASCIIStringHuge encoding (string length {value.Length} > 65535) " + + "is not implemented; add when a real wire field needs it."); + } + + WriteByte(CacheableAsciiString); + WriteUInt16((ushort)value.Length); + foreach (var c in value) + { + _buffer.WriteByte((byte)c); + } + } + + public void WriteJavaModifiedUtf8(string? value) + { + var s = value ?? string.Empty; + + // Pass 1: compute the modified-UTF-8 byte length so we can write the + // u16 length prefix in one shot. We walk per UTF-16 code unit (char); + // surrogate halves naturally fall into the 3-byte branch and a + // supplementary code point ends up as 6 bytes — exactly what Java + // modified UTF-8 calls for. + int byteLen = 0; + foreach (var c in s) + { + if (c >= 0x0001 && c <= 0x007F) + { + byteLen += 1; + } + else if (c == 0 || (c >= 0x0080 && c <= 0x07FF)) + { + byteLen += 2; + } + else + { + byteLen += 3; + } + } + + if (byteLen > 0xFFFF) + { + throw new FormatException( + $"String too long for Java modified UTF-8: {byteLen} bytes (max 65535)."); + } + + WriteUInt16((ushort)byteLen); + + // Pass 2: emit the bytes. + Span buf = stackalloc byte[3]; + foreach (var c in s) + { + if (c >= 0x0001 && c <= 0x007F) + { + _buffer.WriteByte((byte)c); + } + else if (c == 0 || (c >= 0x0080 && c <= 0x07FF)) + { + buf[0] = (byte)(0xC0 | (c >> 6)); + buf[1] = (byte)(0x80 | (c & 0x3F)); + _buffer.Write(buf[..2]); + } + else + { + buf[0] = (byte)(0xE0 | (c >> 12)); + buf[1] = (byte)(0x80 | ((c >> 6) & 0x3F)); + buf[2] = (byte)(0x80 | (c & 0x3F)); + _buffer.Write(buf); + } + } + } /// /// Write a string as UTF-16 big-endian with an i32 byte-length prefix. diff --git a/src/Geode.Client/Protocol/ClientProxyMembershipIdBuilder.cs b/src/Geode.Client/Protocol/ClientProxyMembershipIdBuilder.cs new file mode 100644 index 0000000..5c848a2 --- /dev/null +++ b/src/Geode.Client/Protocol/ClientProxyMembershipIdBuilder.cs @@ -0,0 +1,161 @@ +using System.Net; +using System.Security.Cryptography; +using System.Text; +using Geode.Client.Options; +using Microsoft.Extensions.Options; + +namespace Geode.Client.Protocol; + +/// +/// Generates the inner identity blob of a Geode ClientProxyMembershipID +/// — i.e. the bytes carried in step 6c of the handshake. Mirrors cppcache +/// ClientProxyMembershipIDFactory::create + +/// ClientProxyMembershipID::initObjectVars. +/// +/// +/// +/// The blob is a serialised Java InternalDistributedMember +/// (DataSerializableFixedID = 92), not a serialised +/// ClientProxyMembershipID. The outer ClientProxyMembershipID +/// framing (FixedIDByte + DSFid 38 + identity blob + i32 uniqueId) is +/// added by TcrConnection.HandshakeAsync step 6 — this builder only +/// emits the identity bytes. +/// +/// +/// Should be registered as a singleton (Phase 5 DI). All connections in a +/// process share the same identity — matches cppcache where one factory +/// per process holds a single randString_ reused across every +/// create(). The result is cached after the first +/// call since inputs (hostname, IP, PID, options) are immutable. +/// +/// +internal sealed class ClientProxyMembershipIdBuilder(IOptions options) +{ + // === cppcache hardcoded values (ClientProxyMembershipID.cpp:31-33) ====== + private const byte FixedIdByte = 1; + private const byte InternalDistributedMemberDsfid = 92; + private const sbyte VmKindLoner = 13; + private const int DcPort = 12334; + + /// + /// Process-scoped unique tag, generated once at type-load. cppcache + /// builds this in the factory constructor as + /// "Native_" + 10 random alphanumerics + ProcessId. + /// + private static readonly string s_uniqueTag = GenerateUniqueTag(); + + private readonly GeodeClientOptions _options = options.Value; + + /// + /// Cached identity bytes. Inputs are immutable for the lifetime of this + /// builder, so we compute once and reuse on subsequent calls. + /// + private byte[]? _identity; + + /// + /// Build the identity blob. Idempotent — repeated calls return the same + /// byte array reference. + /// + public byte[] Build() + { + if (_identity is not null) + { + return _identity; + } + + var w = new BigEndianBinaryWriter(); + + // Outer framing: this is a serialised InternalDistributedMember. + w.WriteByte(FixedIdByte); + w.WriteByte(InternalDistributedMemberDsfid); + + // Host address: raw IP bytes (4 for IPv4, 16 for IPv6) prefixed + // with varint length via WriteBytes. + w.WriteBytes(ResolveHostAddress()); + + // SyncCounter — reconnect counter; fresh process = 0. + w.WriteInt32(0); + + // Hostname — DSCode-tagged string (server reads via + // StaticSerialization.readString). + w.WriteString(Dns.GetHostName()); + + // SplitBrainFlag — false. cppcache hardcodes 0 in the relevant ctor. + w.WriteSByte(0); + + // DcPort — distributed-cache port; cppcache hardcodes 12334. + w.WriteInt32(DcPort); + + // vPID — process ID, lets the server distinguish co-tenant clients. + w.WriteInt32(Environment.ProcessId); + + // vmKind = LONER (13) — we are not a Geode peer / locator / admin. + w.WriteSByte(VmKindLoner); + + // RoleArrayLength — no roles. Varint encoding (matches server's + // StaticSerialization.readStringArray length sentinel for empty/null). + w.WriteArrayLen(0); + + // dsName — distributed system name; usually "" for clients. + w.WriteString(_options.Name); + + // uniqueTag — randomly generated per process. + w.WriteString(s_uniqueTag); + + // Durable subscription metadata. Server's MemberIdentifierImpl.toData + // / fromDataPre_GFE_9_0_0_0 reads BOTH unconditionally, so we must + // write them every time: + // - empty string + 300 (server's documented default) for non-durable + // - configured values for durable + // The previous "if (durable) throw; else skip" path corrupted the + // wire because the server then read the trailing Version bytes as + // string contents, hitting "Unknown header byte 0". + var sub = _options.Subscription; + w.WriteString(sub.DurableClientId); + w.WriteInt32((int)sub.DurableTimeout.TotalSeconds); + + // Trailing protocol-version stamp (compressed ordinal). + ProtocolVersion.Current.WriteTo(w); + + _identity = w.ToArray(); + return _identity; + } + + /// + /// Resolve the local hostname's first IP and return its raw bytes + /// (4 for IPv4, 16 for IPv6). Mirrors cppcache's + /// resolver.resolve(hostname, "0") followed by taking the first + /// endpoint's address — no filtering by family. + /// + private static byte[] ResolveHostAddress() + { + var hostname = Dns.GetHostName(); + var addresses = Dns.GetHostAddresses(hostname); + if (addresses.Length == 0) + { + throw new InvalidOperationException( + $"No IP address resolved for local hostname '{hostname}'."); + } + return addresses[0].GetAddressBytes(); + } + + /// + /// Generate the process-scoped unique tag. Format matches cppcache + /// ClientProxyMembershipIDFactory ctor exactly so server-side + /// log scraping / tooling is interchangeable. + /// + private static string GenerateUniqueTag() + { + const string alphabet = + "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_"; + + var sb = new StringBuilder(capacity: 7 + 10 + 10); + sb.Append("Native_"); + for (int i = 0; i < 10; i++) + { + sb.Append(alphabet[RandomNumberGenerator.GetInt32(alphabet.Length)]); + } + sb.Append(Environment.ProcessId); + return sb.ToString(); + } +} diff --git a/src/Geode.Client/Protocol/Operations/PingExtensions.cs b/src/Geode.Client/Protocol/Operations/PingExtensions.cs new file mode 100644 index 0000000..4dc0edd --- /dev/null +++ b/src/Geode.Client/Protocol/Operations/PingExtensions.cs @@ -0,0 +1,56 @@ +namespace Geode.Client.Protocol.Operations; + +/// +/// operation on top of +/// . +/// +/// +/// Lives as an extension method (not a method on +/// ) so the connection class stays focused on +/// transport. When the connection pool lands in Phase 6 the wrapper may +/// move to a pool-aware location; the public call site +/// connection.PingAsync(ct) can stay the same shape. +/// +internal static class PingExtensions +{ + /// + /// Send a (5) and wait for the server's + /// (6). Mirrors cppcache + /// TcrMessagePing. + /// + /// + /// Server returned a other than + /// (e.g. an Exception reply carrying + /// error text in its parts). + /// + /// + /// Ping is a "meta" request with no transaction context, so we send + /// TransactionId = -1 to match cppcache's writeHeader + /// behaviour when no TxState is present. We do not validate + /// the reply's TransactionId echo — a single connection only + /// has one in-flight request at a time, and the server's echo + /// semantics for meta ops are unspecified. + /// + public static async Task PingAsync( + this TcrConnection connection, + CancellationToken cancellationToken = default) + { + // cppcache MetaTransactionId — used for any request that isn't + // part of a Geode transaction. + const int MetaTransactionId = -1; + + var ping = new TcrMessage( + MessageType: MessageType.Ping, + TransactionId: MetaTransactionId, + EarlyAck: 0, + Parts: []); + + var reply = await connection.SendRequestAsync(ping, cancellationToken).ConfigureAwait(false); + if (reply.MessageType != MessageType.Reply) + { + throw new GeodeException( + $"Expected Reply ({(int)MessageType.Reply}) to Ping, got " + + $"{reply.MessageType} ({(int)reply.MessageType})."); + } + } +} diff --git a/src/Geode.Client/Protocol/ProtocolVersion.cs b/src/Geode.Client/Protocol/ProtocolVersion.cs new file mode 100644 index 0000000..0237166 --- /dev/null +++ b/src/Geode.Client/Protocol/ProtocolVersion.cs @@ -0,0 +1,71 @@ +namespace Geode.Client.Protocol; + +/// +/// Geode wire-protocol version ordinal. Mirrors +/// cppcache/src/Version.hpp and Version.cpp. +/// +/// +/// +/// Only the ordinal goes on the wire — major/minor/patch are not part of +/// the handshake (despite what some upstream comments imply). Two encodings: +/// +/// +/// +/// Compressed (default, ordinal ≤ ): +/// 1 byte (i8) carrying the ordinal directly. +/// +/// +/// Uncompressed (ordinal > 127): sentinel byte -1 +/// followed by i16 ordinal (3 bytes total). +/// +/// +/// +/// The uncompressed branch is unreachable today ( = 125) +/// but kept in so we don't get caught out when +/// upstream eventually crosses 127. +/// +/// +internal readonly record struct ProtocolVersion(short Ordinal) +{ + /// + /// The ordinal this client identifies itself as on every handshake. + /// cppcache Version::current() hardcodes 125 (= "Geode 1.14.0" + /// wire protocol); any 1.14+ server is backward-compatible. + /// + /// + /// Bump this only when: + /// + /// We need a feature gated behind a newer ordinal. + /// The new ordinal > 127 — at which point + /// starts taking the uncompressed branch; verify it's correct. + /// + /// + public static ProtocolVersion Current => new(125); + + /// + /// Sentinel byte that signals "uncompressed encoding follows" in the + /// cppcache wire format. Defined as kTokenOrdinal there. + /// + private const sbyte TokenOrdinal = -1; + + /// + /// Append this version to using the cppcache + /// Version::write wire format. + /// + public void WriteTo(BigEndianBinaryWriter writer) + { + if (Ordinal <= sbyte.MaxValue) + { + // Compressed form: ordinal fits in i8, write it directly. + writer.WriteSByte((sbyte)Ordinal); + } + else + { + // Uncompressed form: sentinel byte tells the peer "an i16 + // ordinal follows". Unreachable today (Current = 125), kept + // honest so a future bump past 127 just works. + writer.WriteSByte(TokenOrdinal); + writer.WriteInt16(Ordinal); + } + } +} diff --git a/src/Geode.Client/Protocol/TcrConnection.cs b/src/Geode.Client/Protocol/TcrConnection.cs new file mode 100644 index 0000000..6c95ebc --- /dev/null +++ b/src/Geode.Client/Protocol/TcrConnection.cs @@ -0,0 +1,540 @@ +using System.Buffers.Binary; +using System.IO; +using System.Net.Sockets; +using System.Text; +using Geode.Client.Options; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; + +namespace Geode.Client.Protocol; + +/// +/// One framed TCP connection to a Geode server-cache port (default 40404). +/// Mirrors cppcache/src/TcrConnection.cpp. +/// +internal sealed class TcrConnection( + ILogger logger, + IOptions options, + ClientProxyMembershipIdBuilder membershipIdBuilder) + : IAsyncDisposable +{ + readonly TcpClient _tcpClient = new(); + Stream? _stream; + + // Hold the IOptions handle (not .Value) so callers can re-resolve via + // IOptionsMonitor patterns later if needed. Currently consumed by + // HandshakeAsync step 7 (Subscription.ConflateEvents); Phase 6+ pool / + // TLS / auth code will read further fields. + private readonly IOptions _options = options; + + /// + /// Server's subscription-queue role, captured from the handshake reply + /// byte at step 10. Mirrors cppcache hasServerQueue_ — despite + /// the "has" prefix it's an enum, not a bool: + /// 0 = NON_REDUNDANT_SERVER (no subscription queue) + /// 1 = REDUNDANT_PRIMARY_SERVER (primary HA copy) + /// 2 = REDUNDANT_SECONDARY_SERVER (secondary HA copy) + /// MVP doesn't subscribe, so this is informational; Phase 12+ will + /// branch on it for HA failover. + /// + private byte _hasServerQueue; + + /// + /// Number of events currently buffered in the server's subscription + /// queue for this client, captured from handshake step 11. Mirrors + /// cppcache queueSize_. Non-zero only after a reconnect with + /// durable subscriptions — Phase 12+. Default 0. + /// + private int _queueSize; + + /// + /// Server's member identity (a serialised InternalDistributedMember), + /// captured opaque from handshake step 12. null until the + /// handshake completes. Phase 6 (pool) / 7 (locator) will parse this + /// to attribute connections to the right server. + /// + private byte[]? _serverMember; + + /// + /// Whether the server has delta propagation enabled, captured from + /// handshake step 14. Mirrors cppcache m_deltaEnabledOnServer. + /// MVP doesn't send delta updates; recorded for Phase 12+ delta + /// support so the operation layer can branch on it without redoing + /// the handshake. + /// + private bool _deltaEnabled; + + /// + /// Open a TCP connection to : + /// and run the Geode client-to-server handshake. Mirrors + /// cppcache/src/TcrConnection.cpp::initTcrConnection. + /// + /// + /// + /// On return the connection is ready to send framed Geode messages. + /// Bundling TCP connect + handshake in a single entry point matches + /// cppcache and prevents the easy mistake of forgetting to handshake + /// (server rejects the first non-handshake frame). + /// + /// + /// MVP only opens request/response channels; notification channels + /// (subscription / HA secondary) land in Phase 12+ when + /// 's isClientNotification / + /// isSecondary parameters get plumbed through. + /// + /// + public async Task ConnectAsync(string host, int port, CancellationToken cancellationToken = default) + { + ArgumentException.ThrowIfNullOrEmpty(host); + + // Disable Nagle so a 17-byte Ping flushes immediately instead of + // waiting for buffer fill — cppcache does the same. + _tcpClient.NoDelay = true; + await _tcpClient.ConnectAsync(host, port, cancellationToken).ConfigureAwait(false); + logger.LogDebug("TcrConnection connected to {host}:{port}", host, port); + _stream = _tcpClient.GetStream(); + + // Geode handshake — fail fast here if the server rejects us, so the + // caller never sees a half-initialised connection. + await HandshakeAsync(cancellationToken: cancellationToken).ConfigureAwait(false); + } + + /// + /// Run the Geode client-to-server handshake on the already-connected + /// stream. Mirrors cppcache/src/TcrConnection.cpp::initTcrConnection + /// (the portion after createConnection) and + /// HandShake.cpp. + /// + async Task HandshakeAsync( + bool isClientNotification = false, + bool isSecondary = false, + CancellationToken cancellationToken = default) + { + // cppcache precondition: isSecondary only makes sense on a notification + // channel (it picks PRIMARY vs SECONDARY for HA queue replay). + if (isSecondary && !isClientNotification) + { + throw new ArgumentException( + $"{nameof(isSecondary)} requires {nameof(isClientNotification)} = true.", + nameof(isSecondary)); + } + + // Build the whole client-hello in memory; flushed in one SendAsync + // at the end of the client→server section so the bytes hit the wire + // as a single TCP segment. + var hello = new BigEndianBinaryWriter(); + + // === Client → Server ==================================================== + // + // 1. ConnectionType (u8) + // 100 = CLIENT_TO_SERVER — request / response (Phase 2–11) + // 101 = PRIMARY_SERVER_TO_CLIENT — notification / subscription channel + // 102 = SECONDARY_SERVER_TO_CLIENT — HA secondary (server keeps the + // subscription queue as backup, doesn't actively push) + const byte ClientToServer = 100; + const byte PrimaryServerToClient = 101; + const byte SecondaryServerToClient = 102; + var connectionType = isClientNotification + ? (isSecondary ? SecondaryServerToClient : PrimaryServerToClient) + : ClientToServer; + hello.WriteByte(connectionType); + + // + // 2. ProtocolVersion (ordinal only — major/minor/patch never go on the + // wire). Compressed form: ordinal ≤ 127 → 1 byte. Uncompressed: + // sentinel + i16. See ProtocolVersion.WriteTo. + ProtocolVersion.Current.WriteTo(hello); + logger.LogTrace("TcrConnection handshake, sending ProtocolVersion ordinal {Ordinal}", + ProtocolVersion.Current.Ordinal); + // + // 3. ReplyOk (u8) = 59 + // Tells server we are ready to receive its acceptance reply. + // Defined in cppcache/src/TcrConnection.hpp:41 as + // `#define REPLY_OK 59`. (The inline comment at TcrConnection.cpp:160 + // claims 58 — that comment is stale; the macro value 59 is what + // actually goes on the wire.) + const byte ReplyOk = 59; + hello.WriteByte(ReplyOk); + + // + // 4. Port set — channel-type dependent, NO bytes for request/response. + // cppcache TcrConnection.cpp:161-170: + // - !isClientNotification → record local TCP port into a shared set + // (Geode uses the set later to identify which client a notification + // channel belongs to). NO bytes written here. Skipped entirely until + // Phase 6 (pool) / Phase 12+ (subscriptions) need it. + // - isClientNotification → write i32 PortCount + i32 × N port list. + // Phase 12+. + if (isClientNotification) + { + throw new NotImplementedException( + "Notification-channel handshake (port-set list) is not " + + "implemented; subscription support lands in Phase 12+."); + } + + // + // 5. ReadTimeout (i32) — request/response channel only. + // int.MaxValue - 10000 (~24.85 days, "effectively no timeout"). The + // -10000 dodges an old GFE 5.7 bug where the server added a 5-sec + // buffer that would otherwise overflow int.MaxValue. + // Notification channels skip this field (server is the sender, no + // timeout to set). + if (!isClientNotification) + { + const int HandshakeReadTimeoutMillis = int.MaxValue - 10000; + hello.WriteInt32(HandshakeReadTimeoutMillis); + } + + // + // 6. ClientProxyMembershipID — one DataSerializable object on the wire. + // Java client writes this as `DataSerializer.writeObject(id, out)`; + // server reads it as `ClientProxyMembershipID.readCanonicalized(in)` + // which internally calls `DataSerializer.readObject`. The single + // `writeObject` call expands into FOUR sequential wire pieces: + // + // 6a. FixedIDByte (u8 = 1) ← DataSerializableFixedID byte form + // 6b. DSFid (u8 = 38) ← ClientProxyMembershipId class id + // 6c. identity (varint length + bytes) ← cppcache m_memID; + // opaque blob containing + // a serialised + // InternalDistributedMember + // (hostname, PID, version,…) + // 6d. uniqueId (i32) ← reconnect / sync counter; 1 for fresh client + // + // Constants: cppcache/include/geode/internal/DSCode.hpp:28 + // (FixedIDByte = 1) and DSFixedId.hpp:47 (ClientProxyMembershipId = 38). + // TODO: extract DSCode / DSFid enums once Phase 3+ accumulates values. + // The 6c identity bytes are produced by ClientProxyMembershipIdBuilder + // (mirrors cppcache ClientProxyMembershipIDFactory + initObjectVars). + const byte FixedIdByte = 1; + const byte ClientProxyMembershipIdDsfid = 38; + const int FreshClientUniqueId = 1; + hello.WriteByte(FixedIdByte); // 6a + hello.WriteByte(ClientProxyMembershipIdDsfid); // 6b + hello.WriteBytes(membershipIdBuilder.Build()); // 6c (varint length + bytes) + hello.WriteInt32(FreshClientUniqueId); // 6d + + // + // 7. Overrides (byte[] on the Java side, but always length 1 so far). + // Java: `for (byte b : getOverrides()) hdos.writeByte(b)`. + // Server reads ONE byte: `setOverrides(new byte[] { readByte() })`. + // Currently only conflation override is encoded here, sourced + // from GeodeClientOptions.Subscription.ConflateEvents: + // null → 0 (use server default) + // true → 1 (force conflation on) + // false → 2 (force conflation off) + // TODO: keep an eye on Java geode-core widening this array. + hello.WriteByte(MapConflateEvents()); + + // + // 8. Security mode + optional credentials body. + // SECURITY_CREDENTIALS_NONE = 0 ← MVP + // SECURITY_CREDENTIALS_NORMAL = 1 ← Phase 9 auth + // SECURITY_MULTIUSER_NOTIFICATIONCHANNEL = 3 + // (TcrConnection.hpp:49-51 / Java Handshake.java) + // When mode != NONE, Properties body follows immediately. NONE skips it. + const byte SecurityCredentialsNone = 0; + hello.WriteByte(SecurityCredentialsNone); + + // Flush the whole client-hello in one SendAsync. NoDelay is on + // (set in ConnectAsync), so this lands as a single TCP segment; + // the server reads it as one contiguous handshake. + var clientHello = hello.ToArray(); + logger.LogTrace("TcrConnection sending client-hello ({byteCount} bytes)", clientHello.Length); + await SendAsync(clientHello, cancellationToken).ConfigureAwait(false); + + // + // === Server → Client ==================================================== + // Order taken from ClientSideHandshakeImpl.handshakeWithServer (Java). + // + // 9. AcceptanceCode (u8) + // 59 = OK (Handshake.java:58 REPLY_OK). + // 60 REFUSED / 61 INVALID / 66 AUTH_NOT_REQUIRED — server keeps + // sending steps 10-14. + // 67 SERVER_IS_LOCATOR / 21 SSL_REQUIRED — server stops here, no + // more bytes to read. + // Strategy: throw immediately for the "no more data" codes (matches + // Java client). For other non-OK codes, capture the byte and keep + // reading so step 13's diagnostic text can land in the exception. + const byte ReplyOkServer = 59; + const byte ReplyServerIsLocator = 67; + const byte ReplySslRequired = 21; + var acceptanceCode = (await ReadHandshakeDataAsync(1, cancellationToken) + .ConfigureAwait(false))[0]; + if (acceptanceCode == ReplySslRequired) + { + throw new GeodeException( + "Geode server requires SSL but client connected in plaintext."); + } + if (acceptanceCode == ReplyServerIsLocator) + { + throw new GeodeException( + "Connected port belongs to a Geode locator, not a server. " + + "Use locator-discovery configuration instead of pointing at this address directly."); + } + // Any other non-OK code → defer the throw until after step 13 so we + // can surface the server's diagnostic message in the exception. + // + // 10. EndpointType / ServerQueueStatus (u8). Identifies the server's + // subscription role (NON_REDUNDANT_SERVER / PRIMARY / SECONDARY). + // MVP doesn't subscribe, but we record the value into + // _hasServerQueue so Phase 12+ HA failover can branch on it + // without re-running the handshake. + _hasServerQueue = (await ReadHandshakeDataAsync(1, cancellationToken) + .ConfigureAwait(false))[0]; + logger.LogTrace("TcrConnection handshake hasServerQueue = {hasServerQueue}", _hasServerQueue); + + // + // 11. QueueSize (i32). Number of events currently buffered in the + // server's subscription queue for this client. Non-zero only + // after reconnect with durable subscriptions; recorded into + // _queueSize for Phase 12+ to consume. + var queueSizeBuf = await ReadHandshakeDataAsync(4, cancellationToken) + .ConfigureAwait(false); + _queueSize = BinaryPrimitives.ReadInt32BigEndian(queueSizeBuf); + logger.LogTrace("TcrConnection handshake queueSize = {queueSize}", _queueSize); + // + // 12. ServerMember — varint length + N opaque bytes (the server's + // serialised InternalDistributedMember). Read via + // DataSerializer.readByteArray on the Java side; same encoding + // as our WriteArrayLen / WriteBytes pair. We capture the bytes + // into _serverMember without parsing — Phase 6/7 will decode. + var serverMemberLen = await ReadHandshakeArrayLenAsync(cancellationToken) + .ConfigureAwait(false); + _serverMember = serverMemberLen > 0 + ? await ReadHandshakeDataAsync(serverMemberLen, cancellationToken).ConfigureAwait(false) + : []; + logger.LogTrace("TcrConnection handshake serverMember = {byteCount} bytes", _serverMember.Length); + // + // 13. Message — Java writeUTF format (u16 byte-length + modified UTF-8). + // Server's diagnostic / refusal text; empty on the success path, + // populated on REFUSED / INVALID / AUTH_NOT_REQUIRED / etc. + // Captured into serverMessage and folded into the GeodeException + // thrown after step 14 when AcceptanceCode != REPLY_OK. + // Modified-UTF-8 vs standard UTF-8 only differs at U+0000 and + // supplementary code points; English diagnostic text decodes + // identically with Encoding.UTF8. + var messageLenBuf = await ReadHandshakeDataAsync(2, cancellationToken) + .ConfigureAwait(false); + var messageLen = BinaryPrimitives.ReadUInt16BigEndian(messageLenBuf); + var messageBytes = messageLen > 0 + ? await ReadHandshakeDataAsync(messageLen, cancellationToken).ConfigureAwait(false) + : []; + var serverMessage = Encoding.UTF8.GetString(messageBytes); + logger.LogTrace("TcrConnection handshake serverMessage = '{serverMessage}'", serverMessage); + // + // 14. DeltaEnabled (u8 read as bool: 0 = false, non-zero = true). + // Server's delta-propagation toggle; recorded into _deltaEnabled + // for Phase 12+ to branch on. Not actioned in MVP. + _deltaEnabled = (await ReadHandshakeDataAsync(1, cancellationToken) + .ConfigureAwait(false))[0] != 0; + logger.LogTrace("TcrConnection handshake deltaEnabled = {deltaEnabled}", _deltaEnabled); + + // Deferred from step 9: now that the full server response is drained + // (so the stream is in a clean state for the caller's next move) and + // the diagnostic text from step 13 is in hand, surface any non-OK + // acceptance code as a GeodeException with the message attached. + if (acceptanceCode != ReplyOkServer) + { + var detail = string.IsNullOrEmpty(serverMessage) + ? "(no message)" + : $"\"{serverMessage}\""; + throw new GeodeException( + $"Geode server refused handshake; AcceptanceCode = {acceptanceCode}. Server says: {detail}."); + } + // + // ======================================================================== + // Implementation strategy for the server response: read each field + // off _stream with ReadHandshakeDataAsync + BinaryPrimitives, and + // validate / drain as listed above. + } + + /// + /// Map the tristate + /// to the wire byte used in the handshake "overrides" field. Mirrors + /// cppcache TcrConnection::getOverrides. + /// + private byte MapConflateEvents() => _options.Value.Subscription.ConflateEvents switch + { + null => 0, // CONFLATION_DEFAULT — let the server decide + true => 1, // CONFLATION_ON + false => 2, // CONFLATION_OFF + }; + + /// + /// Read exactly bytes from the underlying + /// stream — the handshake's ad-hoc, non-framed read primitive. Mirrors + /// cppcache TcrConnection::readHandshakeData. + /// + /// + /// + /// already handles partial reads + cancellation, so this helper is just + /// "allocate buffer + read into it" as a single named step that reads + /// well at the call site. + /// + private async Task ReadHandshakeDataAsync( + int byteCount, + CancellationToken cancellationToken) + { + var stream = _stream + ?? throw new InvalidOperationException( + $"{nameof(ConnectAsync)} must be called before reading handshake data."); + + var buf = new byte[byteCount]; + await stream.ReadExactlyAsync(buf, cancellationToken).ConfigureAwait(false); + return buf; + } + + /// + /// Read Geode's variable-length array-length encoding from the stream. + /// Inverse of / + /// cppcache DataInput::readArrayLen: + /// + /// first byte == -1 (0xFF) → -1 (null sentinel). + /// first byte == -2 (0xFE) → u16 length follows (3 bytes total). + /// first byte == -3 (0xFD) → i32 length follows (5 bytes total). + /// otherwise (0–252) → first byte itself is the length. + /// + /// + private async Task ReadHandshakeArrayLenAsync(CancellationToken cancellationToken) + { + var first = (sbyte)(await ReadHandshakeDataAsync(1, cancellationToken) + .ConfigureAwait(false))[0]; + return first switch + { + -1 => -1, + -2 => BinaryPrimitives.ReadUInt16BigEndian( + await ReadHandshakeDataAsync(2, cancellationToken).ConfigureAwait(false)), + -3 => BinaryPrimitives.ReadInt32BigEndian( + await ReadHandshakeDataAsync(4, cancellationToken).ConfigureAwait(false)), + _ => first, + }; + } + + /// + /// Write a fully-encoded frame to the wire and flush. + /// + /// + /// Pure transport: the caller (operation layer) is responsible for + /// producing via + /// or equivalent. Mirrors cppcache/src/TcrConnection.cpp::send. + /// Assumes the connection is already open; + /// throws otherwise. + /// + public async Task SendAsync(ReadOnlyMemory data, CancellationToken cancellationToken = default) + { + var stream = _stream + ?? throw new InvalidOperationException( + $"{nameof(ConnectAsync)} must be called before {nameof(SendAsync)}."); + + logger.LogTrace("TcrConnection sending {ByteCount} bytes", data.Length); + + await stream.WriteAsync(data, cancellationToken).ConfigureAwait(false); + await stream.FlushAsync(cancellationToken).ConfigureAwait(false); + } + + /// + /// Read one framed message from the wire: 17-byte header followed by + /// the MessageLength body bytes the header advertises. + /// + /// + /// The full frame (header + body) as a contiguous byte array, ready to + /// be handed to by the caller. + /// + /// + /// The peer closed the connection before a full frame was received. + /// + /// + /// The header advertises a negative MessageLength. + /// + /// + /// Pure transport: decoding the bytes back into a + /// is the caller's job. Mirrors cppcache/src/TcrConnection.cpp::readMessage. + /// + public async Task ReceiveAsync(CancellationToken cancellationToken = default) + { + var stream = _stream + ?? throw new InvalidOperationException( + $"{nameof(ConnectAsync)} must be called before {nameof(ReceiveAsync)}."); + + // 1. Read the fixed-length header so we know how many body bytes + // to expect. Header offsets: + // 0 i32 MessageType + // 4 i32 MessageLength <- bytes occupied by the Parts payload + // 8 i32 NumParts + // 12 i32 TransactionId + // 16 u8 EarlyAck + var frame = new byte[TcrMessage.HeaderLength]; + await stream + .ReadExactlyAsync(frame.AsMemory(0, TcrMessage.HeaderLength), cancellationToken) + .ConfigureAwait(false); + + var messageLength = BinaryPrimitives.ReadInt32BigEndian(frame.AsSpan(4, sizeof(int))); + if (messageLength < 0) + { + throw new InvalidDataException( + $"Received header advertises negative MessageLength={messageLength}."); + } + + logger.LogTrace( + "TcrConnection received header, MessageLength={messageLength}", messageLength); + + // 2. Grow the buffer and read the parts payload, if any. + if (messageLength > 0) + { + Array.Resize(ref frame, TcrMessage.HeaderLength + messageLength); + await stream + .ReadExactlyAsync( + frame.AsMemory(TcrMessage.HeaderLength, messageLength), cancellationToken) + .ConfigureAwait(false); + } + + return frame; + } + + /// + /// Send a request and read the next framed + /// message from the wire as the reply. The message-level building + /// block on top of / ; + /// every operation (Ping, Put, Get, …) ultimately composes through + /// here. Mirrors cppcache TcrConnection::sendRequest. + /// + /// + /// Pure request-response: assumes one in-flight request per + /// connection. Doesn't interpret the reply — callers branch on + /// themselves (e.g. Reply vs + /// Exception). Phase 6 connection-pool dispatch will lift this to be + /// the only public entry point used by the operation layer. + /// + public async Task SendRequestAsync( + TcrMessage request, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(request); + + await SendAsync(request.Encode(), cancellationToken).ConfigureAwait(false); + var replyBytes = await ReceiveAsync(cancellationToken).ConfigureAwait(false); + return TcrMessage.Decode(replyBytes); + } + + private bool _disposed; + + public async ValueTask DisposeAsync() + { + if (_disposed) + { + return; + } + _disposed = true; + + // Dispose the stream first so any pending async work (e.g. TLS + // close_notify once Phase 8 swaps in SslStream) gets a chance to + // flush; then drop the underlying socket. _stream is null if + // ConnectAsync was never called. + if (_stream is not null) + { + await _stream.DisposeAsync().ConfigureAwait(false); + } + _tcpClient.Dispose(); + } +} diff --git a/tests/Geode.Client.IntegrationTests/Geode.Client.IntegrationTests.csproj b/tests/Geode.Client.IntegrationTests/Geode.Client.IntegrationTests.csproj index 937d63c..ecd5569 100644 --- a/tests/Geode.Client.IntegrationTests/Geode.Client.IntegrationTests.csproj +++ b/tests/Geode.Client.IntegrationTests/Geode.Client.IntegrationTests.csproj @@ -13,6 +13,7 @@ + diff --git a/tests/Geode.Client.IntegrationTests/GeodeFixture.cs b/tests/Geode.Client.IntegrationTests/GeodeFixture.cs index be23667..5657c5b 100644 --- a/tests/Geode.Client.IntegrationTests/GeodeFixture.cs +++ b/tests/Geode.Client.IntegrationTests/GeodeFixture.cs @@ -24,15 +24,23 @@ public sealed class GeodeFixture : IAsyncLifetime public async ValueTask InitializeAsync() { + // The apachegeode/geode image's default entry runs `gfsh`, which + // exits as soon as the supplied -e scripts finish — taking the + // forked locator + server down with it. Wrap in `sh -c "...gfsh -e... && + // tail -f $log"` so the container stays alive (and tails the server + // log to stdout for diagnostics). _container = new ContainerBuilder() .WithImage("apachegeode/geode:latest") .WithPortBinding(10334, true) .WithPortBinding(40404, true) .WithCommand( - "gfsh", "-e", "start locator --name=loc --port=10334", - "-e", "start server --name=srv --server-port=40404", - "-e", "create region --name=test --type=REPLICATE") - .WithWaitStrategy(Wait.ForUnixContainer().UntilPortIsAvailable(10334)) + "sh", "-c", + "gfsh " + + "-e 'start locator --name=loc --port=10334' " + + "-e 'start server --name=srv --server-port=40404' " + + "-e 'create region --name=test --type=REPLICATE' " + + "&& tail -f /srv/srv.log") + .WithWaitStrategy(Wait.ForUnixContainer().UntilPortIsAvailable(40404)) .Build(); await _container.StartAsync(); diff --git a/tests/Geode.Client.IntegrationTests/PingIntegrationTests.cs b/tests/Geode.Client.IntegrationTests/PingIntegrationTests.cs new file mode 100644 index 0000000..2c95423 --- /dev/null +++ b/tests/Geode.Client.IntegrationTests/PingIntegrationTests.cs @@ -0,0 +1,55 @@ +using Geode.Client.Protocol; +using Geode.Client.Protocol.Operations; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Xunit; + +namespace Geode.Client.IntegrationTests; + +/// +/// End-to-end smoke test: TcrConnection.ConnectAsync (TCP + handshake) +/// followed by PingAsync against a real Apache Geode server running in +/// the shared container. This is the moment +/// of truth for Phase 2 — if any byte in the handshake is wrong, the +/// server will refuse the connection here and the assertion / exception +/// tells us what to fix. +/// +[Collection(nameof(GeodeCollection))] +public class PingIntegrationTests(GeodeFixture fx) +{ + /// + /// Generous enough to absorb cold-start IO and image-pull effects on a + /// CI runner; tight enough that a genuine deadlock surfaces in seconds + /// rather than the xUnit-default 10-minute timeout. + /// + private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(30); + + [Fact] + public async Task PingAsync_succeeds_against_real_server() + { + using var cts = new CancellationTokenSource(TestTimeout); + + // Empty config — every GeodeClientOptions field falls back to its + // declared default, which is what the MVP path should require to + // work against a stock Geode server. + var config = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary()) + .Build(); + + await using var services = new ServiceCollection() + .AddLogging() + .AddGeodeClient(config) + .BuildServiceProvider(); + + var connection = services.GetRequiredService(); + + // ConnectAsync bundles TCP connect + Geode handshake. Failure + // here surfaces as GeodeException (server refused) or IOException + // (transport / framing bug). + await connection.ConnectAsync(fx.LocatorHost, fx.ServerPort, cts.Token); + + // Ping a real server-cache; successful return = the server + // accepted the handshake AND replied with MessageType.Reply (6). + await connection.PingAsync(cts.Token); + } +} diff --git a/tests/Geode.Client.Tests/Geode.Client.Tests.csproj b/tests/Geode.Client.Tests/Geode.Client.Tests.csproj index 7e8f6f8..d510c9f 100644 --- a/tests/Geode.Client.Tests/Geode.Client.Tests.csproj +++ b/tests/Geode.Client.Tests/Geode.Client.Tests.csproj @@ -9,6 +9,7 @@ + all diff --git a/tests/Geode.Client.Tests/Protocol/BigEndianBinaryReaderTests.cs b/tests/Geode.Client.Tests/Protocol/BigEndianBinaryReaderTests.cs index d75d8e3..5e68429 100644 --- a/tests/Geode.Client.Tests/Protocol/BigEndianBinaryReaderTests.cs +++ b/tests/Geode.Client.Tests/Protocol/BigEndianBinaryReaderTests.cs @@ -29,6 +29,63 @@ public void ReadInt64_decodes_big_endian_bytes() Assert.Equal(0x0102030405060708L, r.ReadInt64()); } + [Fact] + public void ReadInt16_decodes_big_endian_bytes() + { + var r = new BigEndianBinaryReader(new byte[] { 0x01, 0x02 }); + Assert.Equal(0x0102, r.ReadInt16()); + } + + [Fact] + public void ReadInt16_decodes_negative_value() + { + var r = new BigEndianBinaryReader(new byte[] { 0xFF, 0xFF }); + Assert.Equal((short)-1, r.ReadInt16()); + } + + [Fact] + public void ReadUInt16_decodes_big_endian_bytes() + { + var r = new BigEndianBinaryReader(new byte[] { 0xAB, 0xCD }); + Assert.Equal((ushort)0xABCD, r.ReadUInt16()); + } + + [Fact] + public void ReadUInt32_decodes_big_endian_bytes() + { + var r = new BigEndianBinaryReader(new byte[] { 0xDE, 0xAD, 0xBE, 0xEF }); + Assert.Equal(0xDEADBEEFu, r.ReadUInt32()); + } + + [Fact] + public void ReadUInt64_decodes_big_endian_bytes() + { + var r = new BigEndianBinaryReader(new byte[] + { + 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, + }); + Assert.Equal(0x0102030405060708UL, r.ReadUInt64()); + } + + [Fact] + public void ReadFloat_decodes_IEEE754_big_endian_bytes() + { + // 0x3F800000 → 1.0f. + var r = new BigEndianBinaryReader(new byte[] { 0x3F, 0x80, 0x00, 0x00 }); + Assert.Equal(1.0f, r.ReadFloat()); + } + + [Fact] + public void ReadDouble_decodes_IEEE754_big_endian_bytes() + { + // 0x3FF0000000000000 → 1.0. + var r = new BigEndianBinaryReader(new byte[] + { + 0x3F, 0xF0, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + }); + Assert.Equal(1.0, r.ReadDouble()); + } + [Fact] public void ReadByte_decodes_single_byte() { @@ -36,6 +93,18 @@ public void ReadByte_decodes_single_byte() Assert.Equal(0xAB, r.ReadByte()); } + [Theory] + [InlineData(0x00, (sbyte)0)] + [InlineData(0x01, (sbyte)1)] + [InlineData(0x7F, (sbyte)127)] + [InlineData(0xFF, (sbyte)-1)] + [InlineData(0x80, (sbyte)-128)] + public void ReadSByte_decodes_two_complement_byte(byte raw, sbyte expected) + { + var r = new BigEndianBinaryReader(new byte[] { raw }); + Assert.Equal(expected, r.ReadSByte()); + } + [Theory] [InlineData((byte)0x00, false)] [InlineData((byte)0x01, true)] diff --git a/tests/Geode.Client.Tests/Protocol/BigEndianBinaryWriterTests.cs b/tests/Geode.Client.Tests/Protocol/BigEndianBinaryWriterTests.cs index e567015..cfddaa0 100644 --- a/tests/Geode.Client.Tests/Protocol/BigEndianBinaryWriterTests.cs +++ b/tests/Geode.Client.Tests/Protocol/BigEndianBinaryWriterTests.cs @@ -31,6 +31,68 @@ public void WriteInt64_emits_big_endian_bytes() w.ToArray()); } + [Fact] + public void WriteInt16_emits_big_endian_bytes() + { + var w = new BigEndianBinaryWriter(); + w.WriteInt16(0x0102); + Assert.Equal(new byte[] { 0x01, 0x02 }, w.ToArray()); + } + + [Fact] + public void WriteInt16_emits_negative_value_as_two_complement_big_endian() + { + var w = new BigEndianBinaryWriter(); + w.WriteInt16(-1); + Assert.Equal(new byte[] { 0xFF, 0xFF }, w.ToArray()); + } + + [Fact] + public void WriteUInt16_emits_big_endian_bytes() + { + var w = new BigEndianBinaryWriter(); + w.WriteUInt16(0xABCD); + Assert.Equal(new byte[] { 0xAB, 0xCD }, w.ToArray()); + } + + [Fact] + public void WriteUInt32_emits_big_endian_bytes() + { + var w = new BigEndianBinaryWriter(); + w.WriteUInt32(0xDEADBEEFu); + Assert.Equal(new byte[] { 0xDE, 0xAD, 0xBE, 0xEF }, w.ToArray()); + } + + [Fact] + public void WriteUInt64_emits_big_endian_bytes() + { + var w = new BigEndianBinaryWriter(); + w.WriteUInt64(0x0102030405060708UL); + Assert.Equal( + new byte[] { 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08 }, + w.ToArray()); + } + + [Fact] + public void WriteFloat_emits_IEEE754_big_endian_bytes() + { + // 1.0f → 0x3F800000 in IEEE 754 single precision. + var w = new BigEndianBinaryWriter(); + w.WriteFloat(1.0f); + Assert.Equal(new byte[] { 0x3F, 0x80, 0x00, 0x00 }, w.ToArray()); + } + + [Fact] + public void WriteDouble_emits_IEEE754_big_endian_bytes() + { + // 1.0 → 0x3FF0000000000000 in IEEE 754 double precision. + var w = new BigEndianBinaryWriter(); + w.WriteDouble(1.0); + Assert.Equal( + new byte[] { 0x3F, 0xF0, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00 }, + w.ToArray()); + } + [Fact] public void WriteByte_emits_single_byte() { @@ -39,6 +101,19 @@ public void WriteByte_emits_single_byte() Assert.Equal(new byte[] { 0xAB }, w.ToArray()); } + [Theory] + [InlineData((sbyte)0, 0x00)] + [InlineData((sbyte)1, 0x01)] + [InlineData((sbyte)127, 0x7F)] + [InlineData((sbyte)-1, 0xFF)] + [InlineData((sbyte)-128, 0x80)] + public void WriteSByte_emits_two_complement_byte(sbyte value, byte expected) + { + var w = new BigEndianBinaryWriter(); + w.WriteSByte(value); + Assert.Equal(new byte[] { expected }, w.ToArray()); + } + [Theory] [InlineData(true, 0x01)] [InlineData(false, 0x00)] diff --git a/tests/Geode.Client.Tests/Protocol/ClientProxyMembershipIdBuilderTests.cs b/tests/Geode.Client.Tests/Protocol/ClientProxyMembershipIdBuilderTests.cs new file mode 100644 index 0000000..377b552 --- /dev/null +++ b/tests/Geode.Client.Tests/Protocol/ClientProxyMembershipIdBuilderTests.cs @@ -0,0 +1,274 @@ +using System.Buffers.Binary; +using System.Net; +using System.Text; +using Geode.Client.Options; +using Geode.Client.Protocol; +using Microsoft.Extensions.Options; +using Xunit; +using OptionsFactory = Microsoft.Extensions.Options.Options; + +namespace Geode.Client.Tests.Protocol; + +/// +/// Verifies the byte layout of the identity blob produced by +/// against the schema +/// expected by Java MemberIdentifierImpl.fromDataPre_GFE_9_0_0_0. +/// +/// +/// We can't lock in exact bytes (hostname / IP / PID vary per machine), +/// so the tests parse the blob and assert the structure + recovered +/// values. The parser doubles as a regression detector — if a future +/// change drops or reorders a field, parsing throws or asserts fail. +/// +public class ClientProxyMembershipIdBuilderTests +{ + private static ClientProxyMembershipIdBuilder NewBuilder(GeodeClientOptions? options = null) + => new(OptionsFactory.Create(options ?? new GeodeClientOptions())); + + // ==================================================================== + // Smoke / invariants + // ==================================================================== + + [Fact] + public void Build_returns_non_empty_byte_array() + { + var bytes = NewBuilder().Build(); + Assert.NotEmpty(bytes); + } + + [Fact] + public void Build_is_idempotent_returns_same_array_reference() + { + var b = NewBuilder(); + var first = b.Build(); + var second = b.Build(); + Assert.Same(first, second); + } + + [Fact] + public void Build_starts_with_FixedIdByte_then_InternalDistributedMember_DSFid() + { + var bytes = NewBuilder().Build(); + Assert.Equal(1, bytes[0]); // FixedIDByte + Assert.Equal(92, bytes[1]); // DSFid InternalDistributedMember + } + + // ==================================================================== + // Full structural decode against default options + // ==================================================================== + + [Fact] + public void Build_full_schema_default_options() + { + var parsed = MembershipBlob.Parse(NewBuilder().Build()); + + Assert.Equal(1, parsed.FixedIdByte); + Assert.Equal(92, parsed.Dsfid); + + // IPv4 = 4 bytes, IPv6 = 16 bytes — neither is empty. + Assert.True(parsed.HostAddress.Length is 4 or 16, + $"Expected IPv4 (4) or IPv6 (16) bytes, got {parsed.HostAddress.Length}"); + + Assert.Equal(0, parsed.SyncCounter); + Assert.Equal(Dns.GetHostName(), parsed.Hostname); + Assert.Equal(0, parsed.SplitBrainFlag); + Assert.Equal(12334, parsed.DcPort); // cppcache kDcPort + Assert.Equal(Environment.ProcessId, parsed.VmPid); + Assert.Equal(13, parsed.VmKind); // VmKindLoner + Assert.Equal(0, parsed.RoleArrayLen); + Assert.Equal(string.Empty, parsed.DsName); // GeodeClientOptions.Name default + Assert.StartsWith("Native_", parsed.UniqueTag); + Assert.Matches(@"^Native_[A-Za-z0-9_]{10}\d+$", parsed.UniqueTag); + Assert.Equal(string.Empty, parsed.DurableClientId); // SubscriptionOptions.DurableClientId default + Assert.Equal(300, parsed.DurableTimeoutSeconds); // SubscriptionOptions.DurableTimeout default + Assert.Equal(125, parsed.VersionOrdinal); // ProtocolVersion.Current + } + + // ==================================================================== + // Options propagation + // ==================================================================== + + [Fact] + public void Build_propagates_dsName_from_options() + { + var opts = new GeodeClientOptions { Name = "test-cluster" }; + var parsed = MembershipBlob.Parse(NewBuilder(opts).Build()); + Assert.Equal("test-cluster", parsed.DsName); + } + + [Fact] + public void Build_propagates_durable_client_id_and_timeout() + { + var opts = new GeodeClientOptions(); + opts.Subscription.DurableClientId = "order-svc-1"; + opts.Subscription.DurableTimeout = TimeSpan.FromMinutes(10); + + var parsed = MembershipBlob.Parse(NewBuilder(opts).Build()); + + Assert.Equal("order-svc-1", parsed.DurableClientId); + Assert.Equal(600, parsed.DurableTimeoutSeconds); + } + + [Fact] + public void Build_writes_durable_fields_unconditionally_when_id_is_empty() + { + // Regression guard: the blob must contain durableClientId="" + 300s + // even for non-durable clients. Skipping these fields was the + // initial bug that caused server "Unknown header byte 0". + var parsed = MembershipBlob.Parse(NewBuilder().Build()); + Assert.Equal(string.Empty, parsed.DurableClientId); + Assert.Equal(300, parsed.DurableTimeoutSeconds); + } + + // ==================================================================== + // Process-scoped uniqueTag identity + // ==================================================================== + + [Fact] + public void Build_two_instances_share_the_same_uniqueTag() + { + var a = MembershipBlob.Parse(NewBuilder().Build()); + var b = MembershipBlob.Parse(NewBuilder().Build()); + Assert.Equal(a.UniqueTag, b.UniqueTag); + } + + [Fact] + public void Build_uses_current_process_id() + { + var parsed = MembershipBlob.Parse(NewBuilder().Build()); + Assert.Equal(Environment.ProcessId, parsed.VmPid); + } + + // ==================================================================== + // Parser — walks the blob per the documented schema and asserts + // the cursor consumes the whole input. + // ==================================================================== + + private sealed record MembershipBlob( + byte FixedIdByte, + byte Dsfid, + byte[] HostAddress, + int SyncCounter, + string Hostname, + sbyte SplitBrainFlag, + int DcPort, + int VmPid, + sbyte VmKind, + int RoleArrayLen, + string DsName, + string UniqueTag, + string DurableClientId, + int DurableTimeoutSeconds, + short VersionOrdinal) + { + public static MembershipBlob Parse(ReadOnlySpan bytes) + { + var pos = 0; + var fixedIdByte = bytes[pos++]; + var dsfid = bytes[pos++]; + var hostAddress = ReadBytesVarintPrefixed(bytes, ref pos); + var syncCounter = ReadInt32(bytes, ref pos); + var hostname = ReadString(bytes, ref pos); + var splitBrainFlag = (sbyte)bytes[pos++]; + var dcPort = ReadInt32(bytes, ref pos); + var vmPid = ReadInt32(bytes, ref pos); + var vmKind = (sbyte)bytes[pos++]; + var roleArrayLen = ReadVarintLen(bytes, ref pos); + var dsName = ReadString(bytes, ref pos); + var uniqueTag = ReadString(bytes, ref pos); + var durableClientId = ReadString(bytes, ref pos); + var durableTimeout = ReadInt32(bytes, ref pos); + var versionOrdinal = ReadProtocolVersion(bytes, ref pos); + + Assert.Equal(bytes.Length, pos); + + return new MembershipBlob( + fixedIdByte, dsfid, hostAddress, syncCounter, hostname, + splitBrainFlag, dcPort, vmPid, vmKind, roleArrayLen, + dsName, uniqueTag, durableClientId, durableTimeout, versionOrdinal); + } + + private static int ReadVarintLen(ReadOnlySpan b, ref int pos) + { + var first = (sbyte)b[pos++]; + if (first == -1) return -1; + if (first == -2) + { + var v = BinaryPrimitives.ReadUInt16BigEndian(b[pos..]); + pos += 2; + return v; + } + if (first == -3) + { + var v = BinaryPrimitives.ReadInt32BigEndian(b[pos..]); + pos += 4; + return v; + } + return first; + } + + private static int ReadInt32(ReadOnlySpan b, ref int pos) + { + var v = BinaryPrimitives.ReadInt32BigEndian(b.Slice(pos, 4)); + pos += 4; + return v; + } + + private static byte[] ReadBytesVarintPrefixed(ReadOnlySpan b, ref int pos) + { + var len = ReadVarintLen(b, ref pos); + if (len <= 0) return []; + var result = b.Slice(pos, len).ToArray(); + pos += len; + return result; + } + + private static string ReadString(ReadOnlySpan b, ref int pos) + { + var dsCode = b[pos++]; + return dsCode switch + { + // CacheableASCIIString = 87 → u16 length + ASCII bytes + 87 => ReadAscii(b, ref pos), + // CacheableString = 42 → u16 byte-length + modified UTF-8. + // Tests only feed ASCII strings, so standard UTF-8 decode + // is byte-equivalent for the cases we cover. + 42 => ReadModUtf8AsAscii(b, ref pos), + // CacheableNullString = 69 → no body. + 69 => null!, + _ => throw new InvalidOperationException( + $"Unexpected string DSCode {dsCode} at position {pos - 1}; " + + "either the writer mis-emitted a string or the schema drifted."), + }; + } + + private static string ReadAscii(ReadOnlySpan b, ref int pos) + { + var len = BinaryPrimitives.ReadUInt16BigEndian(b.Slice(pos, 2)); + pos += 2; + var s = Encoding.ASCII.GetString(b.Slice(pos, len)); + pos += len; + return s; + } + + private static string ReadModUtf8AsAscii(ReadOnlySpan b, ref int pos) + { + var byteLen = BinaryPrimitives.ReadUInt16BigEndian(b.Slice(pos, 2)); + pos += 2; + var s = Encoding.UTF8.GetString(b.Slice(pos, byteLen)); + pos += byteLen; + return s; + } + + private static short ReadProtocolVersion(ReadOnlySpan b, ref int pos) + { + var first = (sbyte)b[pos++]; + // Compressed form (ordinal ≤ 127) — single byte. + if (first != -1) return first; + // Uncompressed: sentinel + i16 ordinal. + var ordinal = BinaryPrimitives.ReadInt16BigEndian(b.Slice(pos, 2)); + pos += 2; + return ordinal; + } + } +}