From 3b55d8d2f8822a7843ecadfc185d9c0e2b88f467 Mon Sep 17 00:00:00 2001 From: Ahmed Zamil Date: Sun, 9 Aug 2026 03:54:28 +0300 Subject: [PATCH] Revisions --- Examples/StandaloneWebServer/Program.cs | 5 +- Libraries/Esiur/Core/ExceptionCode.cs | 3 +- Libraries/Esiur/Data/Codec.cs | 4 +- Libraries/Esiur/Data/IDynamicResource.cs | 25 + Libraries/Esiur/Data/RuntimeCaster.cs | 39 ++ Libraries/Esiur/Data/Types/RemoteTypeDef.cs | 7 +- Libraries/Esiur/Esiur.csproj | 2 +- .../Esiur/Net/Packets/EpPacketRequest.cs | 1 + Libraries/Esiur/Protocol/EpConnection.cs | 49 +- .../Esiur/Protocol/EpConnectionProtocol.cs | 471 +++++++++++++++--- Libraries/Esiur/Protocol/EpProtocol.cs | 12 + Libraries/Esiur/Protocol/EpResource.cs | 174 ++++++- .../Esiur/Protocol/EpResourceQueueItem.cs | 16 +- Libraries/Esiur/Protocol/EpServer.cs | 25 +- Libraries/Esiur/Proxy/TypeDefGenerator.cs | 6 +- Libraries/Esiur/README.md | 27 +- .../Esiur/Resource/CustomEventOccurredInfo.cs | 14 +- Libraries/Esiur/Resource/EventOccurredInfo.cs | 17 +- Libraries/Esiur/Resource/Instance.cs | 250 +++++++++- .../Resource/PropertyModificationInfo.cs | 19 +- Libraries/Esiur/Resource/ResourceCursor.cs | 36 ++ Libraries/Esiur/Resource/ResourceJournal.cs | 240 +++++++++ Libraries/Esiur/Resource/Warehouse.cs | 52 +- Libraries/Esiur/Stores/MemoryStore.cs | 28 +- Libraries/Esiur/Stores/TemporaryStore.cs | 18 +- README.md | 61 ++- Stores/Esiur.Stores.EntityCore/EntityStore.cs | 18 +- Tests/Esiur.CLI.Tests/ConfigurationTests.cs | 3 +- .../Esiur.Tests.RPC.EsiurServer.Service.g.cs | 3 +- Tests/Unit/EpPortTests.cs | 20 +- Tests/Unit/Integration/BeaconResource.cs | 4 +- .../EventSubscriptionIntegrationTests.cs | 119 +++++ Tests/Unit/Integration/IntegrationHarness.cs | 26 +- .../SessionHeadersIntegrationTests.cs | 94 ++++ Tests/Unit/RuntimeCasterRecordTests.cs | 44 ++ .../Configuration/ConfigurationResolver.cs | 35 +- Tools/Esiur.CLI/README.md | 6 +- 37 files changed, 1743 insertions(+), 230 deletions(-) create mode 100644 Libraries/Esiur/Protocol/EpProtocol.cs create mode 100644 Libraries/Esiur/Resource/ResourceCursor.cs create mode 100644 Libraries/Esiur/Resource/ResourceJournal.cs create mode 100644 Tests/Unit/RuntimeCasterRecordTests.cs diff --git a/Examples/StandaloneWebServer/Program.cs b/Examples/StandaloneWebServer/Program.cs index ce10c89..bde2a1f 100644 --- a/Examples/StandaloneWebServer/Program.cs +++ b/Examples/StandaloneWebServer/Program.cs @@ -14,15 +14,12 @@ internal class Program private static async Task Main(string[] args) { - var epPort = ushort.Parse( - Environment.GetEnvironmentVariable("ESIUR_PORT") - ?? throw new InvalidOperationException("Set ESIUR_PORT before starting the example.")); var wh = new Warehouse(); // Create a store to keep objects. var system = await wh.Put("sys", new MemoryStore()); // Create a distibuted server - var esiurServer = await wh.Put("sys/server", new EpServer() { Port = epPort }); + var esiurServer = await wh.Put("sys/server", new EpServer()); // Add your object to the store var service = await wh.Put("sys/demo", new Demo()); diff --git a/Libraries/Esiur/Core/ExceptionCode.cs b/Libraries/Esiur/Core/ExceptionCode.cs index 2f095ed..cfc1fab 100644 --- a/Libraries/Esiur/Core/ExceptionCode.cs +++ b/Libraries/Esiur/Core/ExceptionCode.cs @@ -51,5 +51,6 @@ public enum ExceptionCode : ushort ParserLimitExceeded, AttachmentLimitExceeded, AlreadyAttached, - ConnectionLimitExceeded + ConnectionLimitExceeded, + CursorExpired } diff --git a/Libraries/Esiur/Data/Codec.cs b/Libraries/Esiur/Data/Codec.cs index b030e09..757edb5 100644 --- a/Libraries/Esiur/Data/Codec.cs +++ b/Libraries/Esiur/Data/Codec.cs @@ -228,13 +228,13 @@ public static class Codec asyncReply.Then(value => { rt.Trigger(new ParseResult(value, (uint)tdu.TotalLength)); - }); + }).Error(rt.TriggerError); } else { rt.Trigger(new ParseResult(result, (uint)tdu.TotalLength)); } - }); + }).Error(rt.TriggerError); return rt; diff --git a/Libraries/Esiur/Data/IDynamicResource.cs b/Libraries/Esiur/Data/IDynamicResource.cs index 68080ef..c8226db 100644 --- a/Libraries/Esiur/Data/IDynamicResource.cs +++ b/Libraries/Esiur/Data/IDynamicResource.cs @@ -1,5 +1,6 @@ using Esiur.Core; using Esiur.Data.Types; +using Esiur.Protocol; using System; using System.Collections.Generic; using System.Text; @@ -17,4 +18,28 @@ namespace Esiur.Data public TypeDef ResourceDefinition { get; } } + + /// + /// Optional server-side invocation hook for local resources whose type + /// definition is assembled at runtime. Remote + /// proxies remain unchanged; this hook gives their local counterpart the + /// same dynamic function semantics. + /// + public interface IDynamicResourceFunctionHandler + { + public AsyncReply InvokeResourceFunctionAsync( + byte index, + object arguments, + InvocationContext context); + } + + public delegate void DynamicResourceEventHandler(byte index, object value); + + /// + /// Optional event bridge for a local runtime-defined resource. + /// + public interface IDynamicResourceEventSource + { + public event DynamicResourceEventHandler ResourceEventOccurred; + } } diff --git a/Libraries/Esiur/Data/RuntimeCaster.cs b/Libraries/Esiur/Data/RuntimeCaster.cs index 608ecfe..1ad95d7 100644 --- a/Libraries/Esiur/Data/RuntimeCaster.cs +++ b/Libraries/Esiur/Data/RuntimeCaster.cs @@ -5,6 +5,7 @@ using System.Collections.Concurrent; using System.Collections.Generic; using System.Globalization; using System.Linq.Expressions; +using System.Reflection; using System.Runtime.CompilerServices; namespace Esiur.Data; @@ -55,6 +56,9 @@ public static class RuntimeCaster Func> _elemConvCache = new ConcurrentDictionary<(Type, Type), Func>(); + private static readonly ConcurrentDictionary _recordPropertyCache = + new ConcurrentDictionary(); + // --------- Zero-allocation convenience overloads --------- public static object? Cast(object? value, Type toType) => Cast(value, toType, RuntimeCastOptions.Default); @@ -149,6 +153,13 @@ public static class RuntimeCaster var toUnderlying = Nullable.GetUnderlyingType(toType) ?? toType; var fromUnderlying = Nullable.GetUnderlyingType(fromType) ?? fromType; + // A peer without a pre-generated CLR proxy represents a remote typed + // record as Record. Materialize it into the function's declared local + // IRecord type so dynamic resources retain their strongly typed + // invocation contract across the wire. + if (value is Record record && typeof(IRecord).IsAssignableFrom(toUnderlying)) + return ConvertRecord(record, toUnderlying, opts); + // Collections early { bool handled; @@ -233,6 +244,34 @@ public static class RuntimeCaster } } + private static object ConvertRecord(Record record, Type targetType, RuntimeCastOptions opts) + { + if (targetType.IsAbstract || targetType.IsInterface) + throw new InvalidCastException("Cannot materialize a record as " + targetType + "."); + + var target = Activator.CreateInstance(targetType) + ?? throw new InvalidCastException("Cannot create " + targetType + "."); + var properties = _recordPropertyCache.GetOrAdd( + targetType, + static type => type.GetProperties(BindingFlags.Instance | BindingFlags.Public) + .Where(property => + property.CanWrite && + property.SetMethod?.IsPublic == true && + property.GetIndexParameters().Length == 0) + .ToArray()); + + foreach (var property in properties) + { + if (!record.ContainsKey(property.Name)) + continue; + + var converted = Cast(record[property.Name], property.PropertyType, opts); + property.SetValue(target, converted); + } + + return target; + } + private static object? ConvertCollectionsIfAny(object value, Type fromType, Type toType, RuntimeCastOptions opts, out bool handled) { handled = false; diff --git a/Libraries/Esiur/Data/Types/RemoteTypeDef.cs b/Libraries/Esiur/Data/Types/RemoteTypeDef.cs index d70c69d..e1d9c06 100644 --- a/Libraries/Esiur/Data/Types/RemoteTypeDef.cs +++ b/Libraries/Esiur/Data/Types/RemoteTypeDef.cs @@ -189,7 +189,12 @@ public class RemoteTypeDef:TypeDef definition._content = data.Clip(offset, contentLength); definition._typeId = info.Id; definition._parentTypeId = info.Parent; - definition._typeName = string.IsNullOrEmpty(info.Namespace) + // Older TypeDefInfo producers put the fully-qualified name in Name + // while also filling Namespace. Avoid duplicating the namespace, but + // continue accepting producers that send a simple type name. + definition._typeName = string.IsNullOrEmpty(info.Namespace) || + info.Name.StartsWith(info.Namespace + ".", StringComparison.Ordinal) || + info.Name.StartsWith(info.Namespace + "+", StringComparison.Ordinal) ? info.Name : $"{info.Namespace}.{info.Name}"; definition._typeDefKind = info.Kind; diff --git a/Libraries/Esiur/Esiur.csproj b/Libraries/Esiur/Esiur.csproj index 9c770b1..9558523 100644 --- a/Libraries/Esiur/Esiur.csproj +++ b/Libraries/Esiur/Esiur.csproj @@ -5,7 +5,7 @@ Ahmed Kh. Zamil https://www.esiur.com true - 3.0.1 + 3.1.0 https://github.com/esiur/esiur-dotnet Ahmed Kh. Zamil diff --git a/Libraries/Esiur/Net/Packets/EpPacketRequest.cs b/Libraries/Esiur/Net/Packets/EpPacketRequest.cs index 1e01cf4..1ff906c 100644 --- a/Libraries/Esiur/Net/Packets/EpPacketRequest.cs +++ b/Libraries/Esiur/Net/Packets/EpPacketRequest.cs @@ -11,6 +11,7 @@ namespace Esiur.Net.Packets SetProperty = 0x1, Subscribe = 0x2, Unsubscribe = 0x3, + QueryResourceJournal = 0x4, // Request Inquire TypeDefIdsByNames = 0x8, diff --git a/Libraries/Esiur/Protocol/EpConnection.cs b/Libraries/Esiur/Protocol/EpConnection.cs index ea11758..e02d84d 100644 --- a/Libraries/Esiur/Protocol/EpConnection.cs +++ b/Libraries/Esiur/Protocol/EpConnection.cs @@ -135,7 +135,7 @@ public partial class EpConnection : NetworkConnection, IStore long _authenticationAttemptGeneration; string _hostname; - ushort _port; + ushort _port = EpProtocol.DefaultPort; bool _initialPacket = true; AuthenticationDirection _authDirection = AuthenticationDirection.Responder; @@ -456,6 +456,12 @@ public partial class EpConnection : NetworkConnection, IStore var headers = _session.LocalHeaders.Copy(); + // Anonymous sessions still exchange typed records. They therefore + // need a non-empty schema namespace just like authenticated sessions; + // otherwise the responder cannot cache a fetched RemoteTypeDef. + if (string.IsNullOrWhiteSpace(headers.Domain)) + headers.Domain = _remoteDomain ?? _hostname ?? "anonymous"; + if (_session.AuthenticationMode != AuthenticationMode.None) { if (_session.AuthenticationHandler == null) @@ -1215,9 +1221,9 @@ public partial class EpConnection : NetworkConnection, IStore _queue.Then((x) => { if (x.Type == EpResourceQueueItem.DistributedResourceQueueItemType.Event) - x.Resource._EmitEventByIndex(x.Index, x.Value); + x.Resource._EmitEventByIndex(x.Index, x.Value, x.Cursor, x.RecordedAt); else - x.Resource._UpdatePropertyByIndex(x.Index, x.Value); + x.Resource._UpdatePropertyByIndex(x.Index, x.Value, x.Cursor, x.RecordedAt); }).Error(e => { // do nothing @@ -1377,6 +1383,9 @@ public partial class EpConnection : NetworkConnection, IStore case EpPacketRequest.Unsubscribe: EpRequestUnsubscribe(_packet.CallbackId, dt); break; + case EpPacketRequest.QueryResourceJournal: + EpRequestQueryResourceJournal(_packet.CallbackId, dt); + break; // Inquire case EpPacketRequest.TypeDefIdsByNames: EpRequestTypeDefIdsByNames(_packet.CallbackId, dt); @@ -1635,6 +1644,14 @@ public partial class EpConnection : NetworkConnection, IStore } _session.RemoteHeaders = remoteHeaders; + // Responder-side connections do not pass through the + // outbound EpConnectionContext path that initializes + // _remoteDomain. Keep the peer schema namespace from the + // authentication headers so typed records sent from the + // initiator can register and resolve their TypeDefs. + _remoteDomain = string.IsNullOrWhiteSpace(remoteHeaders.Domain) + ? $"anonymous:{RemoteEndPoint?.Address}" + : remoteHeaders.Domain; _session.AuthenticationMode = _authPacket.AuthMode; var localHeaders = _session.LocalHeaders.Copy(); @@ -3144,15 +3161,11 @@ public partial class EpConnection : NetworkConnection, IStore if (!Uri.TryCreate($"ep://{Instance.Name}", UriKind.Absolute, out var endpoint) - || string.IsNullOrWhiteSpace(endpoint.Host) - || endpoint.Port <= 0 - || endpoint.Port > ushort.MaxValue) - throw new FormatException( - "EP endpoints must include an explicit port (for example, ep://host:port)." - ); + || string.IsNullOrWhiteSpace(endpoint.Host)) + throw new FormatException("The EP endpoint is invalid."); var address = endpoint.Host; - var port = checked((ushort)endpoint.Port); + var port = ResolveEndpointPort(endpoint); // assign domain from hostname if not provided if (context is EpConnectionContext epContext) @@ -3198,6 +3211,20 @@ public partial class EpConnection : NetworkConnection, IStore return new AsyncReply(true); } + internal static ushort ResolveEndpointPort(Uri endpoint) + { + if (endpoint == null) + throw new ArgumentNullException(nameof(endpoint)); + + if (endpoint.Port < 0) + return EpProtocol.DefaultPort; + + if (endpoint.Port == 0 || endpoint.Port > ushort.MaxValue) + throw new FormatException("The EP endpoint port must be from 1 through 65535."); + + return checked((ushort)endpoint.Port); + } + @@ -3568,7 +3595,7 @@ public partial class EpConnection : NetworkConnection, IStore // Reattach using the last-known age so only properties modified while // disconnected are transferred and merged, instead of re-fetching all. - await Reattach(r.ResourceLink, r.Instance.Age, r); + await Reattach(r.ResourceLink, r.Instance.Cursor, r); Global.Log("EpConnection", LogType.Debug, "Restored " + r.ResourceInstanceId); diff --git a/Libraries/Esiur/Protocol/EpConnectionProtocol.cs b/Libraries/Esiur/Protocol/EpConnectionProtocol.cs index 50e25ac..d085e8b 100644 --- a/Libraries/Esiur/Protocol/EpConnectionProtocol.cs +++ b/Libraries/Esiur/Protocol/EpConnectionProtocol.cs @@ -530,7 +530,17 @@ partial class EpConnection internal AsyncReply SendSubscribeRequest(uint instanceId, byte index) { - return SendRequest(EpPacketRequest.Subscribe, instanceId, index); + return SendSubscribeRequest(instanceId, index, new ResourceCursor(Guid.Empty, 0)); + } + + internal AsyncReply SendSubscribeRequest(uint instanceId, byte index, ResourceCursor after) + { + return SendRequest( + EpPacketRequest.Subscribe, + instanceId, + index, + after.Generation.ToByteArray(), + after.Revision); } internal AsyncReply SendUnsubscribeRequest(uint instanceId, byte index) @@ -877,12 +887,14 @@ partial class EpConnection void EpNotificationPropertyModified(PlainTdu tdu) { - // resourceId, index, value + // resourceId, generation, revision, recordedAt, index, value var (valueOffset, valueSize, args) = - DataDeserializer.LimitedCountListParser(tdu.Data, tdu.PayloadOffset, tdu.PayloadLength, Instance.Warehouse, 2); + DataDeserializer.LimitedCountListParser(tdu.Data, tdu.PayloadOffset, tdu.PayloadLength, Instance.Warehouse, 5); var rid = Convert.ToUInt32(args[0]); - var index = (byte)args[1]; + var cursor = new ResourceCursor(new Guid((byte[])args[1]), Convert.ToUInt64(args[2])); + var recordedAt = ((DateTime)args[3]).ToUniversalTime(); + var index = (byte)args[4]; FetchResource(rid, null).Then(r => { @@ -902,14 +914,14 @@ partial class EpConnection { item.Trigger(new EpResourceQueueItem((EpResource)r, EpResourceQueueItem.DistributedResourceQueueItemType.Propery, - result, index)); + result, index, cursor, recordedAt)); }); } else { _queue.Add(new AsyncReply(new EpResourceQueueItem((EpResource)r, EpResourceQueueItem.DistributedResourceQueueItemType.Propery, - value, index)), hasResource: false); + value, index, cursor, recordedAt)), hasResource: false); } } @@ -937,13 +949,15 @@ partial class EpConnection void EpNotificationEventOccurred(PlainTdu tdu) { - // resourceId, index, value + // resourceId, generation, revision, recordedAt, index, value var (valueOffset, valueSize, args) = DataDeserializer.LimitedCountListParser(tdu.Data, tdu.PayloadOffset, - tdu.PayloadLength, Instance.Warehouse, 2); + tdu.PayloadLength, Instance.Warehouse, 5); var resourceId = Convert.ToUInt32(args[0]); - var index = (byte)args[1]; + var cursor = new ResourceCursor(new Guid((byte[])args[1]), Convert.ToUInt64(args[2])); + var recordedAt = ((DateTime)args[3]).ToUniversalTime(); + var index = (byte)args[4]; FetchResource(resourceId, null).Then(r => { @@ -964,13 +978,15 @@ partial class EpConnection asyncReply.Then((result) => { item.Trigger(new EpResourceQueueItem((EpResource)r, - EpResourceQueueItem.DistributedResourceQueueItemType.Event, result, index)); + EpResourceQueueItem.DistributedResourceQueueItemType.Event, + result, index, cursor, recordedAt)); }); } else { item.Trigger(new EpResourceQueueItem((EpResource)r, - EpResourceQueueItem.DistributedResourceQueueItemType.Event, pr.Value, index)); + EpResourceQueueItem.DistributedResourceQueueItemType.Event, + pr.Value, index, cursor, recordedAt)); } }).Error((ex) => throw ex); @@ -1019,19 +1035,22 @@ partial class EpConnection var r = res as IResource; - // unsubscribe - Unsubscribe(r); - - // reply ok - SendReply(EpPacketReply.Completed, callback, - r.Instance.Definition.Id, - r.Instance.Age, - r.Instance.Link, - r.Instance.Hops, - r.Instance.Serialize()); - - // subscribe - Subscribe(r); + r.Instance.SynchronizeJournal(head => + { + // Register live delivery before capturing/sending the + // snapshot. Emission shares the journal lock, so no + // revision can fall between snapshot and subscription. + Unsubscribe(r); + Subscribe(r); + SendReply(EpPacketReply.Completed, callback, + r.Instance.Definition.Id, + head.Generation.ToByteArray(), + head.Revision, + r.Instance.Link, + r.Instance.Hops, + r.Instance.Serialize()); + return true; + }); } else { @@ -1056,12 +1075,14 @@ partial class EpConnection // resourceLinkOrId (string link, or uint id — a link is resolved by // query since a remote node's instance id is not permanent: the // resource may have been cleared from memory and recreated under a - // different id while the peer was disconnected), age + // different id while the peer was disconnected), generation, revision var (valueOffset, valueSize, args) = DataDeserializer.LimitedCountListParser(tdu.Data, tdu.PayloadOffset, - tdu.PayloadLength, Instance.Warehouse, 2); + tdu.PayloadLength, Instance.Warehouse, 3); - var age = Convert.ToUInt64(args[1]); + var requestedCursor = new ResourceCursor( + new Guid((byte[])args[1]), + Convert.ToUInt64(args[2])); void Resolved(IResource res) { @@ -1094,22 +1115,28 @@ partial class EpConnection var r = res; - // unsubscribe - Unsubscribe(r); + r.Instance.SynchronizeJournal(head => + { + var reset = requestedCursor.Generation != head.Generation; + Unsubscribe(r); + Subscribe(r); - // reply ok — the resolved id comes first so the caller can - // detect and apply an id change (link-based reattach, or the - // remote node recreated the resource with a new id). - SendReply(EpPacketReply.Completed, callback, - r.Instance.Id, - r.Instance.Definition.Id, - r.Instance.Age, - r.Instance.Link, - r.Instance.Hops, - r.Instance.SerializeAfter(age)); - - // subscribe - Subscribe(r); + // The resolved id comes first so the caller can detect and + // apply an id change. A generation change returns a full + // property snapshot because the old cursor is unrelated. + SendReply(EpPacketReply.Completed, callback, + r.Instance.Id, + r.Instance.Definition.Id, + head.Generation.ToByteArray(), + head.Revision, + r.Instance.Link, + r.Instance.Hops, + reset, + reset + ? r.Instance.SerializeMap() + : r.Instance.SerializeAfter(requestedCursor.Revision)); + return true; + }); } finally { @@ -1427,7 +1454,10 @@ partial class EpConnection var value = Codec.ParseSync(tdu, Instance.Warehouse); - var typeId = Convert.ToUInt32(value); + // TypeDef identifiers are 64-bit stable hashes. Generated definitions + // often happened to fit in UInt32, which hid this truncation until a + // dynamic resource published a full-width identifier. + var typeId = Convert.ToUInt64(value); var t = Instance.Warehouse.GetLocalTypeDefById(typeId); @@ -1665,7 +1695,7 @@ partial class EpConnection var (offset, length, args) = DataDeserializer.LimitedCountListParser(tdu.Data, tdu.PayloadOffset, tdu.PayloadLength, Instance.Warehouse, 2); - var typeId = Convert.ToUInt32(args[0]); + var typeId = Convert.ToUInt64(args[0]); var index = (byte)args[1]; var typeDef = Instance.Warehouse.GetLocalTypeDefById(typeId); @@ -1814,6 +1844,10 @@ partial class EpConnection } }); } + else if (r is IDynamicResourceFunctionHandler dynamicHandler) + { + InvokeDynamicFunction(dynamicHandler, r, ft, callback, result, managerDelay); + } else { InvokeFunction(ft, callback, result, EpPacketRequest.InvokeFunction, managerDelay, r); @@ -1846,13 +1880,61 @@ partial class EpConnection } }); } + else if (r is IDynamicResourceFunctionHandler dynamicHandler) + { + InvokeDynamicFunction(dynamicHandler, r, ft, callback, pr.Value, managerDelay); + } else { InvokeFunction(ft, callback, pr.Value, EpPacketRequest.InvokeFunction, managerDelay, r); } } - }).Error(x => SendError(ErrorType.Management, callback, (ushort)ExceptionCode.ParseError)); ; - }).Error(x => SendError(ErrorType.Management, callback, (ushort)ExceptionCode.ParseError)); ; + }).Error(x => + { + var summary = SummerizeException(x); + SendError(ErrorType.Management, callback, (ushort)ExceptionCode.ParseError, summary.Item2); + }); + }).Error(x => + { + var summary = SummerizeException(x); + SendError(ErrorType.Management, callback, (ushort)ExceptionCode.ParseError, summary.Item2); + }); + } + + void InvokeDynamicFunction( + IDynamicResourceFunctionHandler handler, + IResource resource, + FunctionDef function, + uint callback, + object arguments, + TimeSpan managerDelay) + { + ExecuteRateControlled(callback, managerDelay, () => + { + var context = new InvocationContext(this, callback); + context.BindOperation(resource, function); + try + { + var reply = handler.InvokeResourceFunctionAsync(function.Index, arguments, context); + if (reply == null) + { + SendError(ErrorType.Management, callback, (ushort)ExceptionCode.MethodNotFound); + return; + } + + reply.Then(result => SendReply(EpPacketReply.Completed, callback, result)) + .Error(exception => + { + var (code, message) = SummerizeException(exception); + SendError(ErrorType.Exception, callback, code, message); + }); + } + catch (Exception exception) + { + var (code, message) = SummerizeException(exception); + SendError(ErrorType.Exception, callback, code, message); + } + }); } @@ -2389,6 +2471,9 @@ partial class EpConnection var resourceId = Convert.ToUInt32(args[0]); var index = (byte)args[1]; + var requestedCursor = args.Length >= 4 + ? new ResourceCursor(new Guid((byte[])args[2]), Convert.ToUInt64(args[3])) + : new ResourceCursor(Guid.Empty, 0); Instance.Warehouse.GetById(resourceId).Then((r) => { @@ -2418,14 +2503,13 @@ partial class EpConnection out _)) return; - if (r is EpResource) + if (!IsOperationAllowed(r, et, ActionType.ReceiveEvent)) { - (r as EpResource).Subscribe(et).Then(x => - { - SendReply(EpPacketReply.Completed, callback); - }).Error(x => SendError(ErrorType.Exception, callback, (ushort)ExceptionCode.GeneralFailure)); + SendError(ErrorType.Management, callback, (ushort)ExceptionCode.NotAllowed); + return; } - else + + if (r is EpResource) { lock (_subscriptionsLock) { @@ -2434,17 +2518,97 @@ partial class EpConnection SendError(ErrorType.Management, callback, (ushort)ExceptionCode.NotAttached); return; } - if (_subscriptions[r].Contains(index)) { SendError(ErrorType.Management, callback, (ushort)ExceptionCode.AlreadyListened); return; } - _subscriptions[r].Add(index); - - SendReply(EpPacketReply.Completed, callback); } + + (r as EpResource).Subscribe(et, requestedCursor).Then(x => + { + var head = r.Instance.Cursor; + SendReply(EpPacketReply.Completed, callback, + head.Generation.ToByteArray(), head.Revision); + }).Error(x => + { + lock (_subscriptionsLock) _subscriptions[r].Remove(index); + SendError(ErrorType.Exception, callback, (ushort)ExceptionCode.GeneralFailure); + }); + } + else + { + r.Instance.SynchronizeJournal(head => + { + lock (_subscriptionsLock) + { + if (!_subscriptions.ContainsKey(r)) + { + SendError(ErrorType.Management, callback, (ushort)ExceptionCode.NotAttached); + return false; + } + + if (_subscriptions[r].Contains(index)) + { + SendError(ErrorType.Management, callback, (ushort)ExceptionCode.AlreadyListened); + return false; + } + + _subscriptions[r].Add(index); + } + + var after = requestedCursor.Generation == Guid.Empty + ? head + : requestedCursor; + + if (et.Historical) + { + var replayAfter = after; + while (true) + { + var page = r.Instance.QueryJournal(new ResourceJournalQuery + { + After = replayAfter, + ThroughRevision = head.Revision, + Kind = ResourceJournalEntryKind.EventOccurred, + MemberIndex = index, + Limit = 10000, + }); + + if (page.CursorExpired) + { + lock (_subscriptionsLock) _subscriptions[r].Remove(index); + SendError( + ErrorType.Management, + callback, + (ushort)ExceptionCode.CursorExpired, + $"The requested cursor is older than {page.OldestAvailable}."); + return false; + } + + foreach (var entry in page.Entries) + SendNotification(EpPacketNotification.EventOccurred, + r.Instance.Id, + entry.Cursor.Generation.ToByteArray(), + entry.Cursor.Revision, + entry.RecordedAt, + entry.MemberIndex, + entry.Value); + + if (!page.HasMore) + break; + if (page.Next == replayAfter) + throw new InvalidOperationException( + "The resource journal did not advance while replaying a page."); + replayAfter = page.Next; + } + } + + SendReply(EpPacketReply.Completed, callback, + head.Generation.ToByteArray(), head.Revision); + return true; + }); } }); @@ -2491,6 +2655,9 @@ partial class EpConnection { (r as EpResource).Unsubscribe(et).Then(x => { + lock (_subscriptionsLock) + if (_subscriptions.ContainsKey(r)) + _subscriptions[r].Remove(index); SendReply(EpPacketReply.Completed, callback); }).Error(x => SendError(ErrorType.Exception, callback, (ushort)ExceptionCode.GeneralFailure)); } @@ -2518,6 +2685,141 @@ partial class EpConnection }); } + void EpRequestQueryResourceJournal(uint callback, PlainTdu tdu) + { + // resourceId, generation, afterRevision, throughRevision?, fromTime?, + // toTime?, kind?, memberIndex?, limit + var (_, _, args) = DataDeserializer.LimitedCountListParser( + tdu.Data, + tdu.PayloadOffset, + tdu.PayloadLength, + Instance.Warehouse, + 9); + + var resourceId = Convert.ToUInt32(args[0]); + var query = new ResourceJournalQuery + { + After = new ResourceCursor(new Guid((byte[])args[1]), Convert.ToUInt64(args[2])), + ThroughRevision = args[3] == null ? null : Convert.ToUInt64(args[3]), + FromTime = args[4] as DateTime?, + ToTime = args[5] as DateTime?, + Kind = args[6] == null + ? null + : (ResourceJournalEntryKind?)Convert.ToByte(args[6]), + MemberIndex = args[7] == null ? null : (byte?)Convert.ToByte(args[7]), + Limit = Convert.ToInt32(args[8]), + }; + + Instance.Warehouse.GetById(resourceId).Then(resource => + { + if (resource == null) + { + SendError(ErrorType.Management, callback, (ushort)ExceptionCode.ResourceNotFound); + return; + } + + lock (_subscriptionsLock) + { + if (!_subscriptions.ContainsKey(resource)) + { + SendError(ErrorType.Management, callback, (ushort)ExceptionCode.NotAttached); + return; + } + } + + var page = resource.Instance.QueryJournal(query); + var permitted = page.Entries.Where(entry => + { + MemberDef member; + ActionType action; + if (entry.Kind == ResourceJournalEntryKind.PropertyModified) + { + member = resource.Instance.Definition.GetPropertyDefByIndex(entry.MemberIndex); + action = ActionType.GetProperty; + } + else + { + member = resource.Instance.Definition.GetEventDefByIndex(entry.MemberIndex); + action = ActionType.ReceiveEvent; + } + + return member != null && IsOperationAllowed(resource, member, action); + }).Select(entry => (object)new object[] + { + entry.Cursor.Generation.ToByteArray(), + entry.Cursor.Revision, + entry.RecordedAt, + (byte)entry.Kind, + entry.MemberIndex, + entry.Value, + }).ToArray(); + + SendReply(EpPacketReply.Completed, callback, + page.OldestAvailable.Generation.ToByteArray(), + page.OldestAvailable.Revision, + page.HighWatermark.Generation.ToByteArray(), + page.HighWatermark.Revision, + page.Next.Generation.ToByteArray(), + page.Next.Revision, + page.CursorExpired, + page.HasMore, + permitted); + }).Error(exception => + SendError(ErrorType.Exception, callback, (ushort)ExceptionCode.GeneralFailure, exception.Message)); + } + + /// Queries retained resource changes without changing live subscriptions. + public AsyncReply QueryResourceJournal( + uint resourceId, + ResourceJournalQuery query) + { + query ??= new ResourceJournalQuery(); + var reply = new AsyncReply(); + SendRequest( + EpPacketRequest.QueryResourceJournal, + resourceId, + query.After.Generation.ToByteArray(), + query.After.Revision, + query.ThroughRevision, + query.FromTime, + query.ToTime, + query.Kind.HasValue ? (object)(byte)query.Kind.Value : null, + query.MemberIndex, + query.Limit).Then(result => + { + try + { + var values = (object[])result; + var entries = ((object[])values[8]).Select(raw => + { + var item = (object[])raw; + return new ResourceJournalEntry( + new ResourceCursor(new Guid((byte[])item[0]), Convert.ToUInt64(item[1])), + ((DateTime)item[2]).ToUniversalTime(), + (ResourceJournalEntryKind)Convert.ToByte(item[3]), + Convert.ToByte(item[4]), + item[5]); + }).ToArray(); + + reply.Trigger(new ResourceJournalPage( + new ResourceCursor(new Guid((byte[])values[0]), Convert.ToUInt64(values[1])), + new ResourceCursor(new Guid((byte[])values[2]), Convert.ToUInt64(values[3])), + new ResourceCursor(new Guid((byte[])values[4]), Convert.ToUInt64(values[5])), + Convert.ToBoolean(values[6]), + Convert.ToBoolean(values[7]), + entries)); + } + catch (Exception exception) + { + reply.TriggerError(new AsyncException( + ErrorType.Management, + (ushort)ExceptionCode.ParseError, + exception.Message)); + } + }).Error(reply.TriggerError); + return reply; + } + @@ -3151,13 +3453,15 @@ partial class EpConnection return; } - // TypeId, Age, Link, Hops, PropertyValue[] + // TypeId, Generation, Revision, Link, Hops, PropertyValue[] var args = (object[])result; - var typeId = Convert.ToUInt32(args[0]); - var age = Convert.ToUInt64(args[1]); - var link = (string)args[2]; - var hops = (byte)args[3]; - var pvData = (byte[])args[4]; + var typeId = Convert.ToUInt64(args[0]); + var cursor = new ResourceCursor( + new Guid((byte[])args[1]), + Convert.ToUInt64(args[2])); + var link = (string)args[3]; + var hops = (byte)args[4]; + var pvData = (byte[])args[5]; var typeDef = resource != null ? @@ -3176,7 +3480,7 @@ partial class EpConnection { var pvs = results as PropertyValue[]; - dr._Attach(pvs); + dr._Attach(pvs, cursor); // Progress signal: a resource has fully attached. Used by tests to // distinguish a true deadlock (no progress while requests pend) from // merely slow processing (these counters keep advancing). @@ -3201,9 +3505,9 @@ partial class EpConnection if (resource == null) { if (td.ProxyType != null) - resource = Activator.CreateInstance(td.ProxyType, this, id, Convert.ToUInt64(args[1]), (string)args[2]) as EpResource; + resource = Activator.CreateInstance(td.ProxyType, this, id, cursor.Revision, link) as EpResource; else - resource = new EpResource(this, id, Convert.ToUInt64(args[1]), (string)args[2]); + resource = new EpResource(this, id, cursor.Revision, link); resource.ResourceDefinition = td; typeDef = td; @@ -3226,9 +3530,9 @@ partial class EpConnection if (resource == null) { if (typeDef.ProxyType != null) - resource = Activator.CreateInstance(typeDef.ProxyType, this, id, Convert.ToUInt64(args[1]), (string)args[2]) as EpResource; + resource = Activator.CreateInstance(typeDef.ProxyType, this, id, cursor.Revision, link) as EpResource; else - resource = new EpResource(this, id, Convert.ToUInt64(args[1]), (string)args[2]); + resource = new EpResource(this, id, cursor.Revision, link); resource.ResourceDefinition = typeDef; @@ -3266,8 +3570,8 @@ partial class EpConnection /// DistributedResource /// /// - /// Re-attaches an already-known resource after reconnection using its last-known age. The peer - /// returns only the properties modified after (the delta), which are + /// Re-attaches an already-known resource after reconnection using its last-known cursor. The peer + /// returns only the properties modified after (the delta), which are /// merged into the existing instance instead of re-fetching everything. Falls back to a full /// if there is no prior state to merge into. /// @@ -3277,7 +3581,10 @@ partial class EpConnection /// under a different id while disconnected), but the link is — passing it resolves the current /// id and reattaches in a single round trip instead of a separate GetResourceIdByLink first. /// - public AsyncReply Reattach(object resourceLinkOrId, ulong age, EpResource resource) + public AsyncReply Reattach( + object resourceLinkOrId, + ResourceCursor cursor, + EpResource resource) { // The already-attached / already-in-flight fast paths only apply when we // already know a specific id — a link's current id isn't known until the @@ -3300,7 +3607,11 @@ partial class EpConnection var reply = new AsyncReply(); ResourceAttachRequestCount++; - SendRequest(EpPacketRequest.ReattachResource, resourceLinkOrId, age).Then(result => + SendRequest( + EpPacketRequest.ReattachResource, + resourceLinkOrId, + cursor.Generation.ToByteArray(), + cursor.Revision).Then(result => { if (result == null) { @@ -3309,10 +3620,13 @@ partial class EpConnection return; } - // resolvedId, typeId, age, link, hops, delta(index -> PropertyValue) + // resolvedId, typeId, generation, revision, link, hops, reset, delta var args = (object[])result; var resolvedId = Convert.ToUInt32(args[0]); - var deltaData = (byte[])args[5]; + var remoteCursor = new ResourceCursor( + new Guid((byte[])args[2]), + Convert.ToUInt64(args[3])); + var deltaData = (byte[])args[7]; var sequence = new uint[] { resolvedId }; var oldId = resource.ResourceInstanceId; @@ -3354,7 +3668,7 @@ partial class EpConnection // request that happens to share the reused oldId. } - if (!resource._Reattach(delta)) + if (!resource._Reattach(delta, remoteCursor)) { // No prior state to merge into — perform a full attach instead. _resourceRequests.Remove(resolvedId); @@ -3660,6 +3974,9 @@ partial class EpConnection { SendNotification(EpPacketNotification.PropertyModified, info.Resource.Instance.Id, + info.Cursor.Generation.ToByteArray(), + info.Cursor.Revision, + info.RecordedAt, info.PropertyDef.Index, info.Value); } @@ -3689,6 +4006,9 @@ partial class EpConnection // compose the packet SendNotification(EpPacketNotification.EventOccurred, info.Resource.Instance.Id, + info.Cursor.Generation.ToByteArray(), + info.Cursor.Revision, + info.RecordedAt, info.EventDef.Index, info.Value); } @@ -3714,6 +4034,9 @@ partial class EpConnection // compose the packet SendNotification(EpPacketNotification.EventOccurred, info.Resource.Instance.Id, + info.Cursor.Generation.ToByteArray(), + info.Cursor.Revision, + info.RecordedAt, info.Definition.Index, info.Value); } diff --git a/Libraries/Esiur/Protocol/EpProtocol.cs b/Libraries/Esiur/Protocol/EpProtocol.cs new file mode 100644 index 0000000..cf61718 --- /dev/null +++ b/Libraries/Esiur/Protocol/EpProtocol.cs @@ -0,0 +1,12 @@ +namespace Esiur.Protocol; + +/// +/// Shared defaults for the Esiur EP protocol. +/// +public static class EpProtocol +{ + /// + /// Default native EP port used when a server or endpoint does not specify one. + /// + public const ushort DefaultPort = 51018; +} diff --git a/Libraries/Esiur/Protocol/EpResource.cs b/Libraries/Esiur/Protocol/EpResource.cs index 4e61d11..3d11a30 100644 --- a/Libraries/Esiur/Protocol/EpResource.cs +++ b/Libraries/Esiur/Protocol/EpResource.cs @@ -89,8 +89,16 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn // before sending a Subscribe/Unsubscribe request, since the server errors // (AlreadyListened/AlreadyUnsubscribed) on a redundant one. readonly HashSet _subscribedEvents = new(); + // Last committed occurrence for each event. The sequence belongs to the + // resource-wide journal; separate checkpoints let subscriptions filter + // unrelated members without losing replay position. + readonly Dictionary _eventCursors = new(); // Event indices with a subscription-reconciliation loop currently running. readonly HashSet _reconciling = new(); + // Callers of OnAsync wait here until the shared reconciliation loop has + // confirmed the server-side subscription. This prevents an invocation + // from racing the first Subscribe request on a newly attached proxy. + readonly Dictionary> _subscriptionWaiters = new(); @@ -219,7 +227,7 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn this._age = age; } - internal bool _Attach(PropertyValue[] properties) + internal bool _Attach(PropertyValue[] properties, ResourceCursor cursor) { if (_status == ResourceStatus.Attached) return false; @@ -235,6 +243,11 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn this._properties[i] = properties[i].Value; } + Instance.ObserveRemoteCursor(cursor, DateTime.MinValue); + lock (_eventCursors) + foreach (var eventDef in Instance.Definition.Events) + _eventCursors[eventDef.Index] = cursor; + // trigger holded events/property updates. //foreach (var r in afterAttachmentTriggers) // r.Key.Trigger(r.Value); @@ -255,11 +268,13 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn /// prior state to merge into), in which case the caller should perform a full attach. /// /// Modified properties keyed by their property index. - internal bool _Reattach(Map delta) + internal bool _Reattach(Map delta, ResourceCursor cursor) { if (_properties == null || _events == null) return false; // no prior state — caller should perform a full attach instead. + var generationChanged = Instance.Generation != cursor.Generation; + foreach (var kv in delta) { var index = kv.Key; @@ -271,6 +286,14 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn _properties[index] = kv.Value.Value; } + Instance.ObserveRemoteCursor(cursor, DateTime.MinValue); + if (generationChanged) + { + lock (_eventCursors) + foreach (var eventDef in Instance.Definition.Events) + _eventCursors[eventDef.Index] = cursor; + } + _status = Resource.ResourceStatus.Attached; // A reattach can follow an unexpected disconnect + automatic // reconnect: the server-side subscription state keyed to the old, @@ -282,11 +305,23 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn } - protected internal virtual void _EmitEventByIndex(byte index, object args) + protected internal virtual void _EmitEventByIndex( + byte index, + object args, + ResourceCursor cursor, + DateTime recordedAt) { var et = Instance.Definition.GetEventDefByIndex(index); + lock (_eventCursors) + { + if (_eventCursors.TryGetValue(index, out var previous) && + previous.Generation == cursor.Generation && + previous.Revision >= cursor.Revision) + return; + _eventCursors[index] = cursor; + } _events[index]?.Invoke(this, args); - Instance.EmitResourceEvent(et, args); + Instance.ApplyRemoteEvent(et, args, cursor, recordedAt); DispatchListeners(_eventListeners, index, args); } @@ -346,14 +381,38 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn return rt; } - if (!et.Subscribable) + if (!et.Subscribable && !et.Historical) { var rt = new AsyncReply(); rt.TriggerError(new AsyncException(ErrorType.Management, (ushort)ExceptionCode.NotSubscribable, "")); return rt; } - return _connection.SendSubscribeRequest(_instanceId, et.Index); + ResourceCursor cursor; + lock (_eventCursors) + cursor = _eventCursors.TryGetValue(et.Index, out var known) + ? known + : Instance.Cursor; + return _connection.SendSubscribeRequest(_instanceId, et.Index, cursor); + } + + internal AsyncReply Subscribe(EventDef et, ResourceCursor after) + { + if (et == null) + { + var reply = new AsyncReply(); + reply.TriggerError(new AsyncException( + ErrorType.Management, (ushort)ExceptionCode.MethodNotFound, "")); + return reply; + } + if (!et.Subscribable && !et.Historical) + { + var reply = new AsyncReply(); + reply.TriggerError(new AsyncException( + ErrorType.Management, (ushort)ExceptionCode.NotSubscribable, "")); + return reply; + } + return _connection.SendSubscribeRequest(_instanceId, et.Index, after); } public AsyncReply Subscribe(string eventName) @@ -373,7 +432,7 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn return rt; } - if (!et.Subscribable) + if (!et.Subscribable && !et.Historical) { var rt = new AsyncReply(); rt.TriggerError(new AsyncException(ErrorType.Management, (ushort)ExceptionCode.NotSubscribable, "")); @@ -390,6 +449,10 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn return Unsubscribe(et); } + /// Queries retained property changes and event occurrences. + public AsyncReply QueryJournal(ResourceJournalQuery query = null) => + _connection.QueryResourceJournal(_instanceId, query ?? new ResourceJournalQuery()); + /// /// Listen for a property change (On(":propName", cb)) or an /// exported event (On("eventName", cb)). For events where the @@ -401,20 +464,54 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn /// public EpResource On(string name, Action callback) { + OnAsync(name, callback); + return this; + } + + /// + /// Register a listener and complete only after a subscribable event is + /// active on the remote endpoint. Multiple concurrent callers share the + /// same wire-level subscription reconciliation. + /// + public AsyncReply OnAsync(string name, Action callback) + { + var reply = new AsyncReply(); if (name.StartsWith(":")) { var propertyName = name.Substring(1); var pt = Instance.Definition.GetPropertyDefByName(propertyName) ?? throw new Exception($"Unknown property \"{propertyName}\"."); AddListener(_propertyListeners, pt.Index, callback); - return this; + reply.Trigger(this); + return reply; } var et = Instance.Definition.GetEventDefByName(name) ?? throw new Exception($"Unknown event \"{name}\"."); AddListener(_eventListeners, et.Index, callback); - if (et.Subscribable) ReconcileSubscription(et); - return this; + if (et.Subscribable || et.Historical) + ReconcileSubscription(et, reply); + else + reply.Trigger(this); + return reply; + } + + /// + /// Registers an event listener and replays retained occurrences newer than + /// before continuing with live delivery. + /// + public AsyncReply OnFromAsync(string name, ResourceCursor after, Action callback) + { + var et = Instance.Definition.GetEventDefByName(name) + ?? throw new Exception($"Unknown event \"{name}\"."); + if (!et.Historical) + throw new InvalidOperationException($"Event `{name}` is not historical."); + + lock (_eventCursors) _eventCursors[et.Index] = after; + var reply = new AsyncReply(); + AddListener(_eventListeners, et.Index, callback); + ReconcileSubscription(et, reply); + return reply; } /// Remove a listener registered with . @@ -430,7 +527,7 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn var et = Instance.Definition.GetEventDefByName(name); if (et == null) return this; RemoveListener(_eventListeners, et.Index, callback); - if (et.Subscribable) ReconcileSubscription(et); + if (et.Subscribable || et.Historical) ReconcileSubscription(et); return this; } @@ -486,7 +583,7 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn foreach (var et in Instance.Definition.Events) { - if (!et.Subscribable) continue; + if (!et.Subscribable && !et.Historical) continue; var hasListeners = ListenerCount(_eventListeners, et.Index) > 0 || _events[et.Index] != null; if (hasListeners) ReconcileSubscription(et); } @@ -500,8 +597,21 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn /// once the in-flight request settles, rather than replaying every /// transition. /// - void ReconcileSubscription(EventDef et) + void ReconcileSubscription(EventDef et, AsyncReply waiter = null) { + if (waiter != null) + { + lock (_subscriptionWaiters) + { + if (!_subscriptionWaiters.TryGetValue(et.Index, out var waiters)) + { + waiters = new List(); + _subscriptionWaiters[et.Index] = waiters; + } + waiters.Add(waiter); + } + } + lock (_reconciling) { if (!_reconciling.Add(et.Index)) return; @@ -524,7 +634,7 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn if (desired == actual) { - lock (_reconciling) _reconciling.Remove(et.Index); + CompleteSubscriptionReconciliation(et.Index, null); return; } @@ -539,15 +649,37 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn // Re-check: the desired state may have changed while this request // was in flight (another On()/Off() call came in meanwhile). StepSubscription(et); - }).Error((_) => + }).Error((error) => { // Leave `_subscribedEvents` as-is; the next On()/Off() call that // changes the listener count re-triggers reconciliation, so a // transient failure here just needs another transition to retry. - lock (_reconciling) _reconciling.Remove(et.Index); + CompleteSubscriptionReconciliation(et.Index, error); }); } + void CompleteSubscriptionReconciliation(byte eventIndex, AsyncException error) + { + lock (_reconciling) _reconciling.Remove(eventIndex); + + AsyncReply[] waiters = null; + lock (_subscriptionWaiters) + { + if (_subscriptionWaiters.TryGetValue(eventIndex, out var pending)) + { + waiters = pending.ToArray(); + _subscriptionWaiters.Remove(eventIndex); + } + } + + if (waiters == null) return; + foreach (var waiter in waiters) + { + if (error == null) waiter.Trigger(this); + else waiter.TriggerError(error); + } + } + public override bool TryInvokeMember(InvokeMemberBinder binder, object[] args, out object result) { @@ -664,11 +796,17 @@ public class EpResource : DynamicObject, IResource, INotifyPropertyChanged, IDyn } - internal void _UpdatePropertyByIndex(byte index, object value) + internal void _UpdatePropertyByIndex( + byte index, + object value, + ResourceCursor cursor, + DateTime recordedAt) { var pt = Instance.Definition.GetPropertyDefByIndex(index); + if (!Instance.IsNewerPropertyRevision(index, cursor)) + return; _properties[index] = value; - Instance.EmitModification(pt, value); + Instance.ApplyRemotePropertyModification(pt, value, cursor, recordedAt); DispatchListeners(_propertyListeners, index, value); } diff --git a/Libraries/Esiur/Protocol/EpResourceQueueItem.cs b/Libraries/Esiur/Protocol/EpResourceQueueItem.cs index da7c1c8..77db5aa 100644 --- a/Libraries/Esiur/Protocol/EpResourceQueueItem.cs +++ b/Libraries/Esiur/Protocol/EpResourceQueueItem.cs @@ -27,6 +27,7 @@ using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading.Tasks; +using Esiur.Resource; namespace Esiur.Protocol; public class EpResourceQueueItem @@ -41,13 +42,23 @@ public class EpResourceQueueItem byte index; object value; EpResource resource; + ResourceCursor cursor; + DateTime recordedAt; - public EpResourceQueueItem(EpResource resource, DistributedResourceQueueItemType type, object value, byte index) + public EpResourceQueueItem( + EpResource resource, + DistributedResourceQueueItemType type, + object value, + byte index, + ResourceCursor cursor, + DateTime recordedAt) { this.resource = resource; this.index = index; this.type = type; this.value = value; + this.cursor = cursor; + this.recordedAt = recordedAt; } public EpResource Resource @@ -68,4 +79,7 @@ public class EpResourceQueueItem { get { return value; } } + + public ResourceCursor Cursor => cursor; + public DateTime RecordedAt => recordedAt; } diff --git a/Libraries/Esiur/Protocol/EpServer.cs b/Libraries/Esiur/Protocol/EpServer.cs index afca77e..c57bb0b 100644 --- a/Libraries/Esiur/Protocol/EpServer.cs +++ b/Libraries/Esiur/Protocol/EpServer.cs @@ -44,6 +44,18 @@ namespace Esiur.Protocol; public class EpServer : NetworkServer, IResource { + /// + /// Raised after an accepted connection has completed authentication and is + /// ready for bidirectional resource access. Unlike , + /// this never exposes a half-completed handshake to applications. + /// + public event EpConnection.ReadyEvent ConnectionReady; + + /// + /// Raised when an accepted connection is removed from the server. + /// + public event Action ConnectionDisconnected; + sealed class PeerAttemptWindow { public DateTime StartedUtc; @@ -132,13 +144,14 @@ public class EpServer : NetworkServer, IResource } /// - /// Application-supplied native TCP port. Zero requests an ephemeral port from the OS. + /// Native TCP port. Defaults to ; + /// explicitly set zero to request an ephemeral port from the OS. /// public ushort Port { get; set; - } + } = EpProtocol.DefaultPort; /// /// Controls whether warehouse initialization opens Esiur's native TCP listener. @@ -246,6 +259,7 @@ public class EpServer : NetworkServer, IResource connection.ExceptionLevel = ExceptionLevel; connection.AuthenticationTimeout = AuthenticationTimeout; connection.RestartAuthenticationDeadline(); + connection.OnReady += AcceptedConnectionReady; base.Add(connection); return true; } @@ -261,6 +275,7 @@ public class EpServer : NetworkServer, IResource { try { + connection.OnReady -= AcceptedConnectionReady; base.Remove(connection); } finally @@ -269,6 +284,9 @@ public class EpServer : NetworkServer, IResource } } + private void AcceptedConnectionReady(EpConnection connection) + => ConnectionReady?.Invoke(connection); + private bool TryAdmitConnection(EpConnection connection, out string rejectionReason) { rejectionReason = null; @@ -403,8 +421,7 @@ public class EpServer : NetworkServer, IResource protected override void ClientDisconnected(EpConnection connection) { - //connection.OnReady -= ConnectionReadyEventReceiver; - //Warehouse.Remove(connection); + ConnectionDisconnected?.Invoke(connection); } public KeyList Calls { get; } = new KeyList(); diff --git a/Libraries/Esiur/Proxy/TypeDefGenerator.cs b/Libraries/Esiur/Proxy/TypeDefGenerator.cs index 3b973fb..10b5cb9 100644 --- a/Libraries/Esiur/Proxy/TypeDefGenerator.cs +++ b/Libraries/Esiur/Proxy/TypeDefGenerator.cs @@ -533,7 +533,7 @@ public static class TypeDefGenerator if (typeDef.Events.Length > 0) { - rt.AppendLine("protected override void _EmitEventByIndex(byte index, object args) {"); + rt.AppendLine("protected override void _EmitEventByIndex(byte index, object args, ResourceCursor cursor, DateTime recordedAt) {"); rt.AppendLine("switch (index) {"); var eventsList = new StringBuilder(); @@ -554,7 +554,9 @@ public static class TypeDefGenerator eventsList.AppendLine($"[Export] public event ResourceEventHandler<{etTypeName}> {e.Name};"); } - rt.AppendLine("}}"); + rt.AppendLine("}"); + rt.AppendLine("base._EmitEventByIndex(index, args, cursor, recordedAt);"); + rt.AppendLine("}"); rt.AppendLine(eventsList.ToString()); diff --git a/Libraries/Esiur/README.md b/Libraries/Esiur/README.md index d756aa2..138ee5e 100644 --- a/Libraries/Esiur/README.md +++ b/Libraries/Esiur/README.md @@ -89,17 +89,17 @@ Now we can add our resource to the memory store using `warehouse.Put`. await warehouse.Put("sys/hello", new HelloResource()); ``` -Add an `EpServer` to expose the resource through the Esiur EP protocol. Anonymous access is enabled here only to make the development example easy to run; production applications should configure an authentication provider. +Add an `EpServer` to expose the resource through the Esiur EP protocol. The +server and portless `ep://` client URLs use port `51018` by default; set +`EpServer.Port` or include a URL port only when an application needs an +override. Anonymous access is enabled here only to make the development +example easy to run; production applications should configure an +authentication provider. ```C# -var epPort = ushort.Parse( - Environment.GetEnvironmentVariable("ESIUR_PORT") - ?? throw new InvalidOperationException("Set ESIUR_PORT.")); - await warehouse.Put("sys/server", new EpServer { - Port = epPort, AllowUnauthorizedAccess = true, // Development only. }); ``` @@ -118,15 +118,11 @@ To sum up >using Esiur.Resource; >using Esiur.Stores; > ->var epPort = ushort.Parse( -> Environment.GetEnvironmentVariable("ESIUR_PORT") -> ?? throw new InvalidOperationException("Set ESIUR_PORT.")); >var warehouse = new Warehouse(); >await warehouse.Put("sys", new MemoryStore()); >await warehouse.Put("sys/hello", new HelloResource()); >await warehouse.Put("sys/server", new EpServer >{ -> Port = epPort, > AllowUnauthorizedAccess = true, // Development only. >}); >await warehouse.Open(); @@ -138,8 +134,7 @@ To access our resource remotely, we need to use it's full path including the pro ```C# var warehouse = new Warehouse(); -var epPort = ushort.Parse(Environment.GetEnvironmentVariable("ESIUR_PORT")!); -dynamic res = await warehouse.Get($"ep://localhost:{epPort}/sys/hello"); +dynamic res = await warehouse.Get("ep://localhost/sys/hello"); ``` Now we can invoke the exported functions and read/write properties; @@ -157,8 +152,7 @@ Summing up >using Esiur.Resource; > >var warehouse = new Warehouse(); ->var epPort = ushort.Parse(Environment.GetEnvironmentVariable("ESIUR_PORT")!); ->dynamic res = await warehouse.Get($"ep://localhost:{epPort}/sys/hello"); +>dynamic res = await warehouse.Get("ep://localhost/sys/hello"); > >var reply = await res.SayHi("Hi, I'm calling you from dotnet"); > @@ -176,7 +170,7 @@ Esiur has a self describing feature which comes with every language it supports, After installing the Esiur NuGet package, a new command named ***Get-Types*** is added to the Visual Studio Package Manager Console. It generates client-side classes for robust static typing. ```ps -Get-Types "ep://localhost:$env:ESIUR_PORT/sys/hello" +Get-Types "ep://localhost/sys/hello" ``` This will generate and add wrappers for all types needed by our resource. @@ -184,8 +178,7 @@ This will generate and add wrappers for all types needed by our resource. Allowing us to use ```C# var warehouse = new Warehouse(); -var epPort = ushort.Parse(Environment.GetEnvironmentVariable("ESIUR_PORT")!); -var res = await warehouse.Get($"ep://localhost:{epPort}/sys/hello"); +var res = await warehouse.Get("ep://localhost/sys/hello"); var reply = await res.SayHi("Static typing is better"); Console.WriteLine(reply); ``` diff --git a/Libraries/Esiur/Resource/CustomEventOccurredInfo.cs b/Libraries/Esiur/Resource/CustomEventOccurredInfo.cs index 8390254..b466e3e 100644 --- a/Libraries/Esiur/Resource/CustomEventOccurredInfo.cs +++ b/Libraries/Esiur/Resource/CustomEventOccurredInfo.cs @@ -13,15 +13,27 @@ public class CustomEventOccurredInfo public readonly object Value; public readonly object Issuer; public readonly Func Receivers; + public readonly ResourceCursor Cursor; + public ulong Revision => Cursor.Revision; + public readonly DateTime RecordedAt; public string Name => EventDef.Name; - public CustomEventOccurredInfo(IResource resource, EventDef eventDef, Func receivers, object issuer, object value) + public CustomEventOccurredInfo( + IResource resource, + EventDef eventDef, + Func receivers, + object issuer, + object value, + ResourceCursor cursor, + DateTime recordedAt) { Resource = resource; EventDef = eventDef; Receivers = receivers; Issuer = issuer; Value = value; + Cursor = cursor; + RecordedAt = recordedAt; } } diff --git a/Libraries/Esiur/Resource/EventOccurredInfo.cs b/Libraries/Esiur/Resource/EventOccurredInfo.cs index f2ebfc2..7b4b6d3 100644 --- a/Libraries/Esiur/Resource/EventOccurredInfo.cs +++ b/Libraries/Esiur/Resource/EventOccurredInfo.cs @@ -15,12 +15,27 @@ namespace Esiur.Resource public readonly IResource Resource; public readonly object Value; + public readonly ResourceCursor Cursor; + public ulong Revision => Cursor.Revision; + public readonly DateTime RecordedAt; - public EventOccurredInfo(IResource resource, EventDef eventDef, object value) + public EventOccurredInfo( + IResource resource, + EventDef eventDef, + object value, + ResourceCursor cursor, + DateTime recordedAt) { Resource = resource; Value = value; Definition = eventDef; + Cursor = cursor; + RecordedAt = recordedAt; + } + + public EventOccurredInfo(IResource resource, EventDef eventDef, object value) + : this(resource, eventDef, value, new ResourceCursor(Guid.Empty, 0), DateTime.UtcNow) + { } } } diff --git a/Libraries/Esiur/Resource/Instance.cs b/Libraries/Esiur/Resource/Instance.cs index 84f9bc3..302f982 100644 --- a/Libraries/Esiur/Resource/Instance.cs +++ b/Libraries/Esiur/Resource/Instance.cs @@ -49,6 +49,8 @@ public class Instance List ages = new(); List modificationDates = new(); private ulong instanceAge; + private Guid streamGeneration; + private readonly object journalSync = new(); private byte hops; private DateTime instanceModificationDate; @@ -331,12 +333,34 @@ public class Instance } /// - /// Age of the instance, incremented by 1 in every modification. + /// Legacy alias for the resource-wide revision. It advances for every + /// property modification and event occurrence. /// public ulong Age { - get { return instanceAge; } - internal set { instanceAge = value; } + get { lock (journalSync) return instanceAge; } + internal set { lock (journalSync) instanceAge = value; } + } + + /// + /// Current resource revision. Unlike the legacy property-only age, this + /// advances for every property modification and event occurrence. + /// + public ulong Revision + { + get { lock (journalSync) return instanceAge; } + } + + /// Generation of the current resource change stream. + public Guid Generation + { + get { lock (journalSync) return streamGeneration; } + } + + /// Current replay cursor for the resource. + public ResourceCursor Cursor + { + get { lock (journalSync) return new ResourceCursor(streamGeneration, instanceAge); } } /// @@ -454,6 +478,16 @@ public class Instance return props; } + /// Exports every property keyed by its member index. + public Map SerializeMap() + { + var values = Serialize(); + var map = new Map(); + for (byte index = 0; index < values.Length; index++) + map.Add(index, values[index]); + return map; + } + /// /// If True, the instance can be stored to disk. @@ -477,24 +511,25 @@ public class Instance IResource res; if (this.resource.TryGetTarget(out res)) { - instanceAge++; - var now = DateTime.UtcNow; + lock (journalSync) + { + var cursor = NextCursor(); + var now = DateTime.UtcNow; - ages[pt.Index] = instanceAge; - modificationDates[pt.Index] = now; + ages[pt.Index] = cursor.Revision; + modificationDates[pt.Index] = now; + instanceModificationDate = now; - //if (pt.HasHistory) - //{ - // store.Record(res, pt.Name, value, ages[pt.Index], now); - //} - //else //if (pt.Storage == StorageMode.Recordable) - //{ - store.Modify(res, pt, value, ages[pt.Index], now); - //} + store.Modify(res, pt, value, cursor.Revision, now); + CommitJournal(res, new ResourceJournalEntry( + cursor, + now, + ResourceJournalEntryKind.PropertyModified, + pt.Index, + value), pt.Historical); - //ResourceModified?.Invoke(res, pt.Name, value); - - PropertyModified?.Invoke(new PropertyModificationInfo(res, pt, value, instanceAge)); + PropertyModified?.Invoke(new PropertyModificationInfo(res, pt, value, cursor, now)); + } } } @@ -526,7 +561,21 @@ public class Instance IResource res; if (this.resource.TryGetTarget(out res)) { - CustomEventOccurred?.Invoke(new CustomEventOccurredInfo(res, eventDef, receivers, issuer, value)); + lock (journalSync) + { + var cursor = NextCursor(); + var now = DateTime.UtcNow; + instanceModificationDate = now; + // Receiver predicates are session-specific and cannot be replayed safely. + CommitJournal(res, new ResourceJournalEntry( + cursor, + now, + ResourceJournalEntryKind.EventOccurred, + eventDef.Index, + value), false); + CustomEventOccurred?.Invoke(new CustomEventOccurredInfo( + res, eventDef, receivers, issuer, value, cursor, now)); + } } } @@ -535,7 +584,7 @@ public class Instance IResource res; if (this.resource.TryGetTarget(out res)) { - EventOccurred?.Invoke(new EventOccurredInfo(res, eventDef, value)); + EmitResourceEventCore(res, eventDef, value); } } @@ -545,7 +594,7 @@ public class Instance if (this.resource.TryGetTarget(out res)) { var eventDef = definition.GetEventDefByIndex(eventIndex); - EventOccurred?.Invoke(new EventOccurredInfo(res, eventDef, value)); + EmitResourceEventCore(res, eventDef, value); } } @@ -555,7 +604,141 @@ public class Instance if (this.resource.TryGetTarget(out res)) { var eventDef = definition.GetEventDefByIndex(eventIndex); - CustomEventOccurred?.Invoke(new CustomEventOccurredInfo(res, eventDef, receivers, issuer, value)); + EmitCustomResourceEvent(issuer, receivers, eventDef, value); + } + } + + void EmitResourceEventCore(IResource res, EventDef eventDef, object value) + { + if (eventDef == null) + return; + + lock (journalSync) + { + var cursor = NextCursor(); + var now = DateTime.UtcNow; + instanceModificationDate = now; + CommitJournal(res, new ResourceJournalEntry( + cursor, + now, + ResourceJournalEntryKind.EventOccurred, + eventDef.Index, + value), eventDef.Historical); + EventOccurred?.Invoke(new EventOccurredInfo(res, eventDef, value, cursor, now)); + } + } + + ResourceCursor NextCursor() + { + instanceAge++; + return new ResourceCursor(streamGeneration, instanceAge); + } + + void CommitJournal(IResource res, ResourceJournalEntry entry, bool retain) + { + if (store is IResourceJournalStore journal && + !journal.AppendJournalEntry(res, entry, retain)) + throw new InvalidOperationException( + $"The store rejected resource journal revision {entry.Cursor} for `{Link}`."); + } + + /// Reads retained changes from this resource's owning store. + public ResourceJournalPage QueryJournal(ResourceJournalQuery query) + { + lock (journalSync) + { + if (resource.TryGetTarget(out var res) && store is IResourceJournalStore journal) + { + var page = journal.QueryJournal(res, query ?? new ResourceJournalQuery()); + var cursor = new ResourceCursor(streamGeneration, instanceAge); + if (page.HighWatermark == cursor) + return page; + + return new ResourceJournalPage( + page.OldestAvailable, + cursor, + page.Next, + page.CursorExpired, + page.HasMore, + page.Entries); + } + + var currentCursor = new ResourceCursor(streamGeneration, instanceAge); + return new ResourceJournalPage(currentCursor, currentCursor, query?.After ?? currentCursor, + false, false, Array.Empty()); + } + } + + /// + /// Runs an attach/replay operation against an atomic resource high-watermark. + /// Emission uses the same lock, so notifications after the callback are live + /// changes strictly newer than the captured cursor. + /// + internal T SynchronizeJournal(Func action) + { + lock (journalSync) + return action(new ResourceCursor(streamGeneration, instanceAge)); + } + + internal void ObserveRemoteCursor(ResourceCursor cursor, DateTime recordedAt) + { + lock (journalSync) + { + if (streamGeneration != cursor.Generation) + { + streamGeneration = cursor.Generation; + instanceAge = cursor.Revision; + } + else if (cursor.Revision > instanceAge) + instanceAge = cursor.Revision; + + if (recordedAt > instanceModificationDate) + instanceModificationDate = recordedAt; + } + } + + internal bool IsNewerPropertyRevision(byte index, ResourceCursor cursor) + { + lock (journalSync) + { + if (streamGeneration != cursor.Generation) + return true; + return index >= ages.Count || !ages[index].HasValue || ages[index].Value < cursor.Revision; + } + } + + internal void ApplyRemotePropertyModification( + PropertyDef propertyDef, + object value, + ResourceCursor cursor, + DateTime recordedAt) + { + if (!resource.TryGetTarget(out var res)) + return; + + lock (journalSync) + { + ObserveRemoteCursor(cursor, recordedAt); + ages[propertyDef.Index] = cursor.Revision; + modificationDates[propertyDef.Index] = recordedAt; + PropertyModified?.Invoke(new PropertyModificationInfo( + res, propertyDef, value, cursor, recordedAt)); + } + } + + internal void ApplyRemoteEvent( + EventDef eventDef, + object value, + ResourceCursor cursor, + DateTime recordedAt) + { + if (!resource.TryGetTarget(out var res)) + return; + + lock (journalSync) + { + ObserveRemoteCursor(cursor, recordedAt); + EventOccurred?.Invoke(new EventOccurredInfo(res, eventDef, value, cursor, recordedAt)); } } @@ -755,7 +938,14 @@ public class Instance /// Name of the instance. /// Resource to manage. /// Store responsible for the resource. - public Instance(Warehouse warehouse, uint id, string name, IResource resource, IStore store, ulong age = 0) + public Instance( + Warehouse warehouse, + uint id, + string name, + IResource resource, + IStore store, + ulong age = 0, + string resourceKey = null) { this.Warehouse = warehouse; this.store = store; @@ -763,6 +953,17 @@ public class Instance this.id = id; this.name = name ?? ""; this.instanceAge = age; + this.streamGeneration = Guid.NewGuid(); + + if (store is IResourceJournalStore journal) + { + var cursor = journal.OpenJournal( + resource, + resourceKey ?? name ?? string.Empty, + new ResourceCursor(streamGeneration, instanceAge)); + streamGeneration = cursor.Generation; + instanceAge = Math.Max(instanceAge, cursor.Revision); + } //this.attributes = new KeyList(this); //children = new AutoList(this); @@ -779,6 +980,7 @@ public class Instance if (resource is IDynamicResource dynamicResource) { this.definition = dynamicResource.ResourceDefinition; + warehouse.RegisterDynamicTypeDef(this.definition); } else { @@ -796,6 +998,8 @@ public class Instance // connect events if (!(resource is EpResource)) { + if (resource is IDynamicResourceEventSource dynamicEventSource) + dynamicEventSource.ResourceEventOccurred += EmitResourceEventByIndex; Type t = ResourceProxy.GetBaseType(resource); diff --git a/Libraries/Esiur/Resource/PropertyModificationInfo.cs b/Libraries/Esiur/Resource/PropertyModificationInfo.cs index bb28714..5015ca4 100644 --- a/Libraries/Esiur/Resource/PropertyModificationInfo.cs +++ b/Libraries/Esiur/Resource/PropertyModificationInfo.cs @@ -12,15 +12,30 @@ public struct PropertyModificationInfo public readonly PropertyDef PropertyDef; public string Name => PropertyDef.Name; public readonly ulong Age; + public ulong Revision => Cursor.Revision; + public readonly ResourceCursor Cursor; + public readonly DateTime RecordedAt; public object Value; - public PropertyModificationInfo(IResource resource, PropertyDef propertyDef, object value, ulong age) + public PropertyModificationInfo( + IResource resource, + PropertyDef propertyDef, + object value, + ResourceCursor cursor, + DateTime recordedAt) { Resource = resource; PropertyDef = propertyDef; - Age = age; + Cursor = cursor; + Age = cursor.Revision; + RecordedAt = recordedAt; Value = value; } + + public PropertyModificationInfo(IResource resource, PropertyDef propertyDef, object value, ulong age) + : this(resource, propertyDef, value, new ResourceCursor(Guid.Empty, age), DateTime.UtcNow) + { + } } diff --git a/Libraries/Esiur/Resource/ResourceCursor.cs b/Libraries/Esiur/Resource/ResourceCursor.cs new file mode 100644 index 0000000..f23d64a --- /dev/null +++ b/Libraries/Esiur/Resource/ResourceCursor.cs @@ -0,0 +1,36 @@ +using System; + +namespace Esiur.Resource; + +/// +/// Identifies an exact position in a resource's ordered change stream. +/// Generation changes when the stream is recreated; revision increases for +/// every observable property modification or event occurrence. +/// +public readonly struct ResourceCursor : IEquatable +{ + public Guid Generation { get; } + public ulong Revision { get; } + + public ResourceCursor(Guid generation, ulong revision) + { + Generation = generation; + Revision = revision; + } + + public bool IsEmpty => Generation == Guid.Empty && Revision == 0; + + public bool Equals(ResourceCursor other) => + Generation == other.Generation && Revision == other.Revision; + + public override bool Equals(object obj) => + obj is ResourceCursor other && Equals(other); + + public override int GetHashCode() => + (Generation.GetHashCode() * 397) ^ Revision.GetHashCode(); + + public override string ToString() => $"{Generation:N}:{Revision}"; + + public static bool operator ==(ResourceCursor left, ResourceCursor right) => left.Equals(right); + public static bool operator !=(ResourceCursor left, ResourceCursor right) => !left.Equals(right); +} diff --git a/Libraries/Esiur/Resource/ResourceJournal.cs b/Libraries/Esiur/Resource/ResourceJournal.cs new file mode 100644 index 0000000..aeb9365 --- /dev/null +++ b/Libraries/Esiur/Resource/ResourceJournal.cs @@ -0,0 +1,240 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Runtime.CompilerServices; + +namespace Esiur.Resource; + +public enum ResourceJournalEntryKind : byte +{ + PropertyModified = 0, + EventOccurred = 1, +} + +/// An ordered, optionally retained resource change. +public sealed class ResourceJournalEntry +{ + public ResourceCursor Cursor { get; } + public DateTime RecordedAt { get; } + public ResourceJournalEntryKind Kind { get; } + public byte MemberIndex { get; } + public object Value { get; } + + public ResourceJournalEntry( + ResourceCursor cursor, + DateTime recordedAt, + ResourceJournalEntryKind kind, + byte memberIndex, + object value) + { + Cursor = cursor; + RecordedAt = recordedAt.Kind == DateTimeKind.Utc + ? recordedAt + : recordedAt.ToUniversalTime(); + Kind = kind; + MemberIndex = memberIndex; + Value = value; + } +} + +/// Filters a bounded resource-journal query. +public sealed class ResourceJournalQuery +{ + public ResourceCursor After { get; set; } + public ulong? ThroughRevision { get; set; } + public DateTime? FromTime { get; set; } + public DateTime? ToTime { get; set; } + public ResourceJournalEntryKind? Kind { get; set; } + public byte? MemberIndex { get; set; } + public int Limit { get; set; } = 1000; +} + +/// A journal page and the stream boundaries used to read it. +public sealed class ResourceJournalPage +{ + public ResourceCursor OldestAvailable { get; } + public ResourceCursor HighWatermark { get; } + public ResourceCursor Next { get; } + public bool CursorExpired { get; } + public bool HasMore { get; } + public ResourceJournalEntry[] Entries { get; } + + public ResourceJournalPage( + ResourceCursor oldestAvailable, + ResourceCursor highWatermark, + ResourceCursor next, + bool cursorExpired, + bool hasMore, + ResourceJournalEntry[] entries) + { + OldestAvailable = oldestAvailable; + HighWatermark = highWatermark; + Next = next; + CursorExpired = cursorExpired; + HasMore = hasMore; + Entries = entries ?? Array.Empty(); + } +} + +/// +/// Optional store capability for persisting historical property changes and +/// event occurrences. The resource instance owns ordering; the store owns +/// retention and persistence. +/// +public interface IResourceJournalStore +{ + ResourceCursor OpenJournal( + IResource resource, + string resourceKey, + ResourceCursor proposedCursor); + bool AppendJournalEntry(IResource resource, ResourceJournalEntry entry, bool retain); + ResourceJournalPage QueryJournal(IResource resource, ResourceJournalQuery query); + void RemoveJournal(IResource resource); +} + +/// +/// Bounded in-memory implementation for volatile stores and tests. Persistent +/// stores can implement with their own +/// transactional backend while preserving the same protocol contract. +/// +public sealed class ResourceJournalBuffer : IResourceJournalStore +{ + sealed class State + { + public ResourceCursor Head; + public ulong DiscardedThroughRevision; + public readonly LinkedList Entries = new(); + } + + sealed class ResourceReferenceComparer : IEqualityComparer + { + public bool Equals(IResource x, IResource y) => ReferenceEquals(x, y); + public int GetHashCode(IResource obj) => RuntimeHelpers.GetHashCode(obj); + } + + readonly object sync = new(); + readonly Dictionary states = + new(new ResourceReferenceComparer()); + readonly int maximumEntriesPerResource; + + public ResourceJournalBuffer(int maximumEntriesPerResource = 10000) + { + if (maximumEntriesPerResource < 1) + throw new ArgumentOutOfRangeException(nameof(maximumEntriesPerResource)); + this.maximumEntriesPerResource = maximumEntriesPerResource; + } + + public ResourceCursor OpenJournal( + IResource resource, + string resourceKey, + ResourceCursor proposedCursor) + { + if (resource == null) throw new ArgumentNullException(nameof(resource)); + lock (sync) + { + if (states.TryGetValue(resource, out var existing)) + return existing.Head; + + var generation = proposedCursor.Generation == Guid.Empty + ? Guid.NewGuid() + : proposedCursor.Generation; + var cursor = new ResourceCursor(generation, proposedCursor.Revision); + states.Add(resource, new State { Head = cursor }); + return cursor; + } + } + + public bool AppendJournalEntry(IResource resource, ResourceJournalEntry entry, bool retain) + { + if (resource == null) throw new ArgumentNullException(nameof(resource)); + if (entry == null) throw new ArgumentNullException(nameof(entry)); + + lock (sync) + { + if (!states.TryGetValue(resource, out var state)) + { + state = new State { Head = new ResourceCursor(entry.Cursor.Generation, 0) }; + states.Add(resource, state); + } + + if (state.Head.Generation != entry.Cursor.Generation || + entry.Cursor.Revision <= state.Head.Revision) + return false; + + state.Head = entry.Cursor; + if (retain) + state.Entries.AddLast(entry); + + while (state.Entries.Count > maximumEntriesPerResource) + { + state.DiscardedThroughRevision = Math.Max( + state.DiscardedThroughRevision, + state.Entries.First.Value.Cursor.Revision); + state.Entries.RemoveFirst(); + } + + return true; + } + } + + public ResourceJournalPage QueryJournal(IResource resource, ResourceJournalQuery query) + { + if (resource == null) throw new ArgumentNullException(nameof(resource)); + query ??= new ResourceJournalQuery(); + var limit = Math.Max(1, Math.Min(query.Limit, 10000)); + + lock (sync) + { + var proposed = new ResourceCursor(Guid.NewGuid(), 0); + if (!states.TryGetValue(resource, out var state)) + { + state = new State { Head = proposed }; + states.Add(resource, state); + } + + var after = query.After; + var generationMismatch = after.Generation != Guid.Empty && + after.Generation != state.Head.Generation; + var expired = generationMismatch || + (!generationMismatch && + after.Revision < state.DiscardedThroughRevision); + + if (generationMismatch) + after = new ResourceCursor(state.Head.Generation, 0); + + IEnumerable filtered = state.Entries + .Where(entry => entry.Cursor.Revision > after.Revision); + + if (query.ThroughRevision.HasValue) + filtered = filtered.Where(entry => entry.Cursor.Revision <= query.ThroughRevision.Value); + if (query.FromTime.HasValue) + filtered = filtered.Where(entry => entry.RecordedAt >= query.FromTime.Value.ToUniversalTime()); + if (query.ToTime.HasValue) + filtered = filtered.Where(entry => entry.RecordedAt <= query.ToTime.Value.ToUniversalTime()); + if (query.Kind.HasValue) + filtered = filtered.Where(entry => entry.Kind == query.Kind.Value); + if (query.MemberIndex.HasValue) + filtered = filtered.Where(entry => entry.MemberIndex == query.MemberIndex.Value); + + var selected = filtered.Take(limit + 1).ToArray(); + var hasMore = selected.Length > limit; + var entries = hasMore ? selected.Take(limit).ToArray() : selected; + var next = entries.Length == 0 ? after : entries[entries.Length - 1].Cursor; + var oldest = state.Entries.First?.Value.Cursor ?? state.Head; + + return new ResourceJournalPage( + oldest, + state.Head, + next, + expired, + hasMore, + entries); + } + } + + public void RemoveJournal(IResource resource) + { + if (resource == null) return; + lock (sync) states.Remove(resource); + } +} diff --git a/Libraries/Esiur/Resource/Warehouse.cs b/Libraries/Esiur/Resource/Warehouse.cs index cd45a20..8b77d86 100644 --- a/Libraries/Esiur/Resource/Warehouse.cs +++ b/Libraries/Esiur/Resource/Warehouse.cs @@ -1036,7 +1036,14 @@ public class Warehouse var resourceId = (uint)Interlocked.Increment(ref _resourceCounter); - resource.Instance = new Instance(this, resourceId, instanceName, resource, store, resourceContext?.Age ?? 0); + resource.Instance = new Instance( + this, + resourceId, + instanceName, + resource, + store, + resourceContext?.Age ?? 0, + string.Join("/", location)); resource.Instance.Managers.AddRange(resourceManagers); @@ -1310,6 +1317,48 @@ public class Warehouse } } + /// + /// Registers a definition supplied by an + /// without replacing its stable identifier. Dynamic definitions are not + /// backed by a CLR , so they cannot use the + /// incremental local registration path, but remote peers must still be + /// able to resolve them through TypeDefById. + /// + internal void RegisterDynamicTypeDef(TypeDef typeDef) + { + if (typeDef == null) + throw new ArgumentNullException(nameof(typeDef)); + + lock (_typeDefsLock) + { + // A remote definition keeps the producer's id for wire lookups, but + // TryRegisterRemoteTypeDef gives it a Warehouse-local id because remote + // resource/record/enum id spaces can overlap. Dynamic proxy instances + // must reuse that local registration instead of indexing by the wire id. + var registrationId = typeDef is RemoteTypeDef remoteTypeDef + && remoteTypeDef.LocalTypeDefId != 0 + ? remoteTypeDef.LocalTypeDefId + : typeDef.Id; + var existing = _localTypeDefs[registrationId]; + if (existing == null) + { + _localTypeDefs[registrationId] = typeDef; + return; + } + + if (ReferenceEquals(existing, typeDef) || + (existing.Kind == typeDef.Kind && + existing.Version == typeDef.Version && + string.Equals(existing.Name, typeDef.Name, StringComparison.Ordinal))) + { + return; + } + + throw new InvalidOperationException( + $"Dynamic TypeDef id {registrationId} is already registered by '{existing.Name}', not '{typeDef.Name}'."); + } + } + internal KeyList> GetProxyTypesByDomain(string domain) { return _proxyTypeDefs[domain]; @@ -1317,6 +1366,7 @@ public class Warehouse public bool TryRegisterRemoteTypeDef(string domain, RemoteTypeDef typeDef) { + domain = string.IsNullOrWhiteSpace(domain) ? "anonymous" : domain; lock (_typeDefsLock) { if (!_remoteTypeDefs.ContainsKey(domain)) diff --git a/Libraries/Esiur/Stores/MemoryStore.cs b/Libraries/Esiur/Stores/MemoryStore.cs index 7b25fcd..0034340 100644 --- a/Libraries/Esiur/Stores/MemoryStore.cs +++ b/Libraries/Esiur/Stores/MemoryStore.cs @@ -10,7 +10,7 @@ using Esiur.Data.Types; namespace Esiur.Stores; -public class MemoryStore : IStore +public class MemoryStore : IStore, IResourceJournalStore { public Instance Instance { get; set; } @@ -18,6 +18,31 @@ public class MemoryStore : IStore KeyList resources = new KeyList(); + readonly IResourceJournalStore journal; + + public MemoryStore() + : this(new ResourceJournalBuffer()) + { + } + + public MemoryStore(IResourceJournalStore journal) + { + this.journal = journal ?? throw new ArgumentNullException(nameof(journal)); + } + + public ResourceCursor OpenJournal( + IResource resource, + string resourceKey, + ResourceCursor proposedCursor) => + journal.OpenJournal(resource, resourceKey, proposedCursor); + + public bool AppendJournalEntry(IResource resource, ResourceJournalEntry entry, bool retain) => + journal.AppendJournalEntry(resource, entry, retain); + + public ResourceJournalPage QueryJournal(IResource resource, ResourceJournalQuery query) => + journal.QueryJournal(resource, query); + + public void RemoveJournal(IResource resource) => journal.RemoveJournal(resource); public void Destroy() { @@ -161,6 +186,7 @@ public class MemoryStore : IStore AsyncReply IStore.Remove(IResource resource) { resources.Remove(resource.Instance.Id); + journal.RemoveJournal(resource); return AsyncReply.FromResult(true); } diff --git a/Libraries/Esiur/Stores/TemporaryStore.cs b/Libraries/Esiur/Stores/TemporaryStore.cs index 7a5aa2b..6056723 100644 --- a/Libraries/Esiur/Stores/TemporaryStore.cs +++ b/Libraries/Esiur/Stores/TemporaryStore.cs @@ -9,13 +9,28 @@ using Esiur.Data; using Esiur.Data.Types; namespace Esiur.Stores; -public class TemporaryStore : IStore +public class TemporaryStore : IStore, IResourceJournalStore { public Instance Instance { get; set; } public event DestroyedEvent OnDestroy; Dictionary resources = new Dictionary(); + readonly ResourceJournalBuffer journal = new(); + + public ResourceCursor OpenJournal( + IResource resource, + string resourceKey, + ResourceCursor proposedCursor) => + journal.OpenJournal(resource, resourceKey, proposedCursor); + + public bool AppendJournalEntry(IResource resource, ResourceJournalEntry entry, bool retain) => + journal.AppendJournalEntry(resource, entry, retain); + + public ResourceJournalPage QueryJournal(IResource resource, ResourceJournalQuery query) => + journal.QueryJournal(resource, query); + + public void RemoveJournal(IResource resource) => journal.RemoveJournal(resource); public void Destroy() { @@ -90,6 +105,7 @@ public class TemporaryStore : IStore AsyncReply IStore.Remove(IResource resource) { resources.Remove(resource.Instance.Id); + journal.RemoveJournal(resource); return new AsyncReply(true); } diff --git a/README.md b/README.md index d0be2ab..4ce37aa 100644 --- a/README.md +++ b/README.md @@ -173,15 +173,12 @@ The `ep://` URL identifies the logical connection and resource path; `WebSocketUri` identifies the transport endpoint. Esiur automatically requests the case-sensitive `EP` WebSocket subprotocol. -For native TCP, omit `WebSocketUri` and include the EP port in the logical URL: +For native TCP, omit `WebSocketUri`. Esiur uses port `51018` when the logical +URL omits a port: ```csharp -var epPort = ushort.Parse( - Environment.GetEnvironmentVariable("ESIUR_PORT") - ?? throw new InvalidOperationException("Set ESIUR_PORT.")); - dynamic counter = await client.Get( - $"ep://localhost:{epPort}/sys/counter"); + "ep://localhost/sys/counter"); ``` ## Standalone hosting @@ -195,16 +192,12 @@ using Esiur.Protocol; using Esiur.Resource; using Esiur.Stores; -var epPort = ushort.Parse( - Environment.GetEnvironmentVariable("ESIUR_PORT") - ?? throw new InvalidOperationException("Set ESIUR_PORT.")); var warehouse = new Warehouse(); await warehouse.Put("sys", new MemoryStore()); await warehouse.Put("sys/counter", new CounterResource()); await warehouse.Put("sys/server", new EpServer { - Port = epPort, AllowUnauthorizedAccess = true, // Development only. }); @@ -224,6 +217,50 @@ Use a separate `Warehouse` for each isolated server or client runtime. Avoid relying on the static default Warehouse in applications that host more than one Esiur environment. +## Resource revisions and historical events + +Esiur 3 uses one ordered cursor for every observable change made by a resource: + +```text +(generation UUID, resource-wide revision) +``` + +Property modifications and event occurrences share this sequence. Notifications +also carry their original UTC recording time. A new generation means that the +producer cannot continue the previous sequence, so consumers must replace their +snapshot instead of comparing revisions from the two generations. + +Mark an event with `[Historical]` when disconnected consumers must be able to +replay it. Ordinary events remain live-only and do not grow the retained journal. + +```csharp +[Export, Historical] +public event ResourceEventHandler? Reading; + +var checkpoint = remote.Instance.Cursor; +await remote.OnFromAsync("Reading", checkpoint, value => +{ + var reading = (ReadingRecord)value; + // Persist the reading, then save the cursor received through + // remote.Instance.EventOccurred. +}); +``` + +`EpResource.QueryJournal(...)` supports bounded queries by cursor, time, entry +kind, and member. `OnFromAsync(...)` atomically replays retained occurrences and +then continues with live delivery. Each consumer owns its checkpoint; the source +does not delete an event merely because one client received it. Reconnect uses +the per-event checkpoint automatically, and returns `CursorExpired` if retention +can no longer satisfy the requested position. + +The resource's owning store controls retention through `IResourceJournalStore`. +The bundled memory, temporary, and current EntityCore adapters use a bounded +10,000-entry process-local journal per resource. A deployment requiring replay +across producer restarts must use a durable implementation of that interface. + +This changes the EP v3 attach, reattach, subscribe, notification, and journal +request shapes. All communicating v3 peers must be upgraded together. + ## Authentication and encryption Anonymous access is opt-in. Production services should omit `AllowAnonymous`, @@ -350,8 +387,8 @@ EP is self-describing, so clients can work dynamically or generate strongly typed models. Install the v3 CLI as a .NET tool: ```shell -dotnet tool install --global Esiur.CLI --version 3.0.0 -esiur get-template "ep://localhost:${ESIUR_PORT}/sys/counter" --dir Generated +dotnet tool install --global Esiur.CLI --version 3.1.0 +esiur get-template "ep://localhost/sys/counter" --dir Generated ``` Use `--async-setters` to generate asynchronous property setters. The CLI also diff --git a/Stores/Esiur.Stores.EntityCore/EntityStore.cs b/Stores/Esiur.Stores.EntityCore/EntityStore.cs index 1566918..4b75d59 100644 --- a/Stores/Esiur.Stores.EntityCore/EntityStore.cs +++ b/Stores/Esiur.Stores.EntityCore/EntityStore.cs @@ -37,7 +37,7 @@ using System.Collections; using Esiur.Data.Types; namespace Esiur.Stores.EntityCore; -public class EntityStore : IStore +public class EntityStore : IStore, IResourceJournalStore { public Instance Instance { get; set; } @@ -50,6 +50,21 @@ public class EntityStore : IStore Dictionary> DB = new Dictionary>(); object DBLock = new object(); + readonly ResourceJournalBuffer journal = new(); + + public ResourceCursor OpenJournal( + IResource resource, + string resourceKey, + ResourceCursor proposedCursor) => + journal.OpenJournal(resource, resourceKey, proposedCursor); + + public bool AppendJournalEntry(IResource resource, ResourceJournalEntry entry, bool retain) => + journal.AppendJournalEntry(resource, entry, retain); + + public ResourceJournalPage QueryJournal(IResource resource, ResourceJournalQuery query) => + journal.QueryJournal(resource, query); + + public void RemoveJournal(IResource resource) => journal.RemoveJournal(resource); Dictionary TypesByName = new Dictionary(); internal Dictionary TypesByType = new Dictionary(); @@ -293,6 +308,7 @@ public class EntityStore : IStore if (DB[type].ContainsKey(eid)) { DB[type].Remove(eid); + journal.RemoveJournal(resource); return new AsyncReply(true); } } diff --git a/Tests/Esiur.CLI.Tests/ConfigurationTests.cs b/Tests/Esiur.CLI.Tests/ConfigurationTests.cs index 7ef19c0..ec92de7 100644 --- a/Tests/Esiur.CLI.Tests/ConfigurationTests.cs +++ b/Tests/Esiur.CLI.Tests/ConfigurationTests.cs @@ -8,6 +8,7 @@ namespace Esiur.CLI.Tests; public sealed class ConfigurationTests { [Theory] + [InlineData("ep://localhost", "ep://localhost")] [InlineData("ep://localhost:65535", "ep://localhost:65535")] [InlineData("ep://example.test:9000/sys/service", "ep://example.test:9000")] public void EndpointParserExtractsConnectionEndpoint(string value, string expected) => @@ -15,7 +16,7 @@ public sealed class ConfigurationTests [Theory] [InlineData("http://localhost")] - [InlineData("ep://localhost")] + [InlineData("ep://localhost:0")] [InlineData("ep:///missing-host")] [InlineData("not-an-endpoint")] public void EndpointParserRejectsInvalidEndpoints(string value) => diff --git a/Tests/RPC/Client/Model/Esiur/Esiur.Tests.RPC.EsiurServer.Service.g.cs b/Tests/RPC/Client/Model/Esiur/Esiur.Tests.RPC.EsiurServer.Service.g.cs index 7f7e12d..43b9a69 100644 --- a/Tests/RPC/Client/Model/Esiur/Esiur.Tests.RPC.EsiurServer.Service.g.cs +++ b/Tests/RPC/Client/Model/Esiur/Esiur.Tests.RPC.EsiurServer.Service.g.cs @@ -183,12 +183,13 @@ namespace Esiur.Tests.RPC.EsiurServer get => (object)_properties[1]; set => SetResourceProperty(1, value); } - protected override void _EmitEventByIndex(byte index, object args) + protected override void _EmitEventByIndex(byte index, object args, ResourceCursor cursor, DateTime recordedAt) { switch (index) { case 0: MessageUpdated?.Invoke((byte[])args); break; } + base._EmitEventByIndex(index, args, cursor, recordedAt); } [Export] public event ResourceEventHandler MessageUpdated; diff --git a/Tests/Unit/EpPortTests.cs b/Tests/Unit/EpPortTests.cs index 88c383d..7407bf1 100644 --- a/Tests/Unit/EpPortTests.cs +++ b/Tests/Unit/EpPortTests.cs @@ -6,17 +6,19 @@ namespace Esiur.Tests.Unit; public sealed class EpPortTests { [Fact] - public void ServerDoesNotAssignAProtocolPort() - => Assert.Equal(0, new EpServer().Port); + public void ServerUsesTheProtocolDefaultPort() + => Assert.Equal(EpProtocol.DefaultPort, new EpServer().Port); [Fact] - public async Task ClientEndpointWithoutPortIsRejected() + public void ClientEndpointWithoutPortUsesTheProtocolDefault() { - var warehouse = new Warehouse(); - - var exception = await Assert.ThrowsAsync(async () => - await warehouse.Get("ep://localhost/sys/resource")); - - Assert.Contains("explicit port", exception.Message, StringComparison.OrdinalIgnoreCase); + var endpoint = new Uri("ep://localhost/sys/resource"); + Assert.Equal(EpProtocol.DefaultPort, EpConnection.ResolveEndpointPort(endpoint)); } + + [Fact] + public void ExplicitClientPortStillOverridesTheProtocolDefault() + => Assert.Equal( + 12345, + EpConnection.ResolveEndpointPort(new Uri("ep://localhost:12345/sys/resource"))); } diff --git a/Tests/Unit/Integration/BeaconResource.cs b/Tests/Unit/Integration/BeaconResource.cs index 5679e04..cea730a 100644 --- a/Tests/Unit/Integration/BeaconResource.cs +++ b/Tests/Unit/Integration/BeaconResource.cs @@ -9,7 +9,9 @@ public partial class BeaconResource // No [AutoDelivery]: subscribable by default — requires an explicit // Subscribe request before occurrences flow to a given connection. - [Export] public event ResourceEventHandler? Ping; + [Export] + [Historical] + public event ResourceEventHandler? Ping; // Opts out via [AutoDelivery]: flows to every attached connection // unconditionally. diff --git a/Tests/Unit/Integration/EventSubscriptionIntegrationTests.cs b/Tests/Unit/Integration/EventSubscriptionIntegrationTests.cs index 44f76b9..efe13a4 100644 --- a/Tests/Unit/Integration/EventSubscriptionIntegrationTests.cs +++ b/Tests/Unit/Integration/EventSubscriptionIntegrationTests.cs @@ -6,6 +6,110 @@ namespace Esiur.Tests.Unit.Integration; [Collection("Integration")] public class EventSubscriptionIntegrationTests { + [Fact] + public async Task PropertyAndEventNotifications_ShareOneOrderedResourceRevision() + { + await using var cluster = await StartClusterAsync(out var getBeacon).WaitAsync(TimeSpan.FromSeconds(10)); + var remote = await GetRemote(cluster); + var beacon = getBeacon(); + var propertyRevisions = new List(); + var eventRevisions = new List(); + + remote.Instance.PropertyModified += info => propertyRevisions.Add(info.Cursor); + remote.Instance.EventOccurred += info => eventRevisions.Add(info.Cursor); + await remote.OnAsync("Ping", _ => { }); + + beacon.Fire("ping", "ordered"); + await WaitUntilAsync( + () => propertyRevisions.Count == 1 && eventRevisions.Count == 1, + TimeSpan.FromSeconds(3)); + + Assert.Equal(propertyRevisions[0].Generation, eventRevisions[0].Generation); + Assert.Equal(propertyRevisions[0].Revision + 1, eventRevisions[0].Revision); + Assert.Equal(eventRevisions[0], remote.Instance.Cursor); + } + + [Fact] + public async Task QueryJournal_ReturnsHistoricalEventAfterCursor() + { + await using var cluster = await StartClusterAsync(out var getBeacon).WaitAsync(TimeSpan.FromSeconds(10)); + var remote = await GetRemote(cluster); + var beacon = getBeacon(); + var attachedAt = remote.Instance.Cursor; + var ping = remote.Instance.Definition.GetEventDefByName("Ping"); + + beacon.Fire("ping", "retained"); + + var page = await remote.QueryJournal(new ResourceJournalQuery + { + After = attachedAt, + Kind = ResourceJournalEntryKind.EventOccurred, + MemberIndex = ping.Index, + }); + + var entry = Assert.Single(page.Entries); + Assert.False(page.CursorExpired); + Assert.Equal(ResourceJournalEntryKind.EventOccurred, entry.Kind); + Assert.Equal(ping.Index, entry.MemberIndex); + Assert.Equal("retained", entry.Value); + Assert.True(entry.Cursor.Revision > attachedAt.Revision); + } + + [Fact] + public async Task HistoricalSubscription_ReplaysOccurrenceMissedDuringReconnect() + { + await using var cluster = await StartClusterAsync(out var getBeacon).WaitAsync(TimeSpan.FromSeconds(10)); + cluster.Connection.AutoReconnect = true; + cluster.Connection.ReconnectInterval = 1; + var remote = await GetRemote(cluster); + var beacon = getBeacon(); + var received = new List(); + + await remote.OnAsync("Ping", value => received.Add((string)value)); + beacon.Fire("ping", "before"); + await WaitUntilAsync(() => received.Count == 1, TimeSpan.FromSeconds(3)); + + foreach (var serverConnection in cluster.Server.Connections.ToArray()) + serverConnection.Destroy(); + + await WaitUntilAsync(() => !cluster.Connection.IsConnected, TimeSpan.FromSeconds(3)); + beacon.Fire("ping", "while-offline"); + await WaitUntilAsync(() => cluster.Connection.IsConnected, TimeSpan.FromSeconds(5)); + await WaitUntilAsync(() => received.Count == 2, TimeSpan.FromSeconds(5)); + + Assert.Equal(new[] { "before", "while-offline" }, received); + } + + [Fact] + public async Task HistoricalSubscription_ReplaysEveryPageBeforeLiveDelivery() + { + const int occurrenceCount = 10_001; + BeaconResource? beacon = null; + await using var cluster = await IntegrationCluster.StartAsync( + async warehouse => + { + beacon = new BeaconResource(); + await warehouse.Put("sys/beacon", beacon); + }, + resourceJournalCapacity: occurrenceCount + 1).WaitAsync(TimeSpan.FromSeconds(10)); + + for (var index = 0; index < occurrenceCount; index++) + beacon!.Fire("ping", index.ToString()); + + var remote = await GetRemote(cluster); + var received = new List(occurrenceCount); + await remote.OnFromAsync( + "Ping", + new ResourceCursor(remote.Instance.Generation, 0), + value => received.Add((string)value)); + + await WaitUntilAsync( + () => received.Count == occurrenceCount, + TimeSpan.FromSeconds(10)); + Assert.Equal("0", received[0]); + Assert.Equal((occurrenceCount - 1).ToString(), received[^1]); + } + [Fact] public async Task On_DeliversAutoDeliveredEventWithNoSubscribeNeeded() { @@ -60,6 +164,21 @@ public class EventSubscriptionIntegrationTests Assert.Equal(new[] { "x", "y" }, b); // neither received "z" } + [Fact] + public async Task OnAsync_CompletesAfterWireSubscriptionIsReady() + { + await using var cluster = await StartClusterAsync(out var getBeacon).WaitAsync(TimeSpan.FromSeconds(10)); + var remote = await GetRemote(cluster); + var beacon = getBeacon(); + var received = new List(); + + await remote.OnAsync("Ping", value => received.Add((string)value)); + beacon.Fire("ping", "immediate"); + + await WaitUntilAsync(() => received.Count == 1, TimeSpan.FromSeconds(3)); + Assert.Equal(new[] { "immediate" }, received); + } + [Fact] public async Task NativePlusEquals_AutoSubscribesAndUnsubscribes() { diff --git a/Tests/Unit/Integration/IntegrationHarness.cs b/Tests/Unit/Integration/IntegrationHarness.cs index 185f7a3..df58c34 100644 --- a/Tests/Unit/Integration/IntegrationHarness.cs +++ b/Tests/Unit/Integration/IntegrationHarness.cs @@ -205,7 +205,11 @@ internal sealed class IntegrationCluster : IAsyncDisposable bool mismatchedSessionKeys = false, bool requireKeyRotation = false, bool allowAuthentication = true, - bool registerServerAuthenticationProvider = true) + bool registerServerAuthenticationProvider = true, + bool anonymous = false, + int resourceJournalCapacity = 10000, + Action serverCreated = null, + Func populateClient = null) { var port = NextAvailablePort(); @@ -217,10 +221,13 @@ internal sealed class IntegrationCluster : IAsyncDisposable if (encrypted || requireEncryption) serverWh.RegisterEncryptionProvider(new AesEncryptionProvider()); - await serverWh.Put("sys", new MemoryStore()); + await serverWh.Put( + "sys", + new MemoryStore(new ResourceJournalBuffer(resourceJournalCapacity))); var server = await serverWh.Put("sys/server", new EpServer { Port = (ushort)port, + AllowUnauthorizedAccess = anonymous, AllowedAuthenticationProviders = allowAuthentication ? new[] { @@ -234,6 +241,7 @@ internal sealed class IntegrationCluster : IAsyncDisposable : Array.Empty(), RequireEncryption = requireEncryption, }); + serverCreated?.Invoke(server); await populate(serverWh); @@ -246,6 +254,8 @@ internal sealed class IntegrationCluster : IAsyncDisposable mismatchedSessionKeys ? (byte)0x80 : (byte)0, requireKeyRotation) : new TestClientAuthProvider()); + if (populateClient is not null) + await populateClient(cluster.ClientWarehouse); if (encrypted) cluster.ClientWarehouse.RegisterEncryptionProvider(new AesEncryptionProvider()); @@ -255,12 +265,16 @@ internal sealed class IntegrationCluster : IAsyncDisposable $"ep://localhost:{port}", new EpConnectionContext { - AuthenticationMode = AuthenticationMode.InitializerIdentity, - Identity = "tester", + AuthenticationMode = anonymous + ? AuthenticationMode.None + : AuthenticationMode.InitializerIdentity, + Identity = anonymous ? null : "tester", AuthenticationProtocol = oneStepAuthentication ? "one-step" - : PasswordAuthenticationProvider.ProtocolName, - Domain = "test", + : anonymous + ? null + : PasswordAuthenticationProvider.ProtocolName, + Domain = anonymous ? null : "test", EncryptionMode = encrypted ? encryptionMode : EncryptionMode.None, diff --git a/Tests/Unit/Integration/SessionHeadersIntegrationTests.cs b/Tests/Unit/Integration/SessionHeadersIntegrationTests.cs index fbf8225..54b7ccb 100644 --- a/Tests/Unit/Integration/SessionHeadersIntegrationTests.cs +++ b/Tests/Unit/Integration/SessionHeadersIntegrationTests.cs @@ -3,13 +3,77 @@ namespace Esiur.Tests.Unit.Integration; using Esiur.Data; using Esiur.Net.Sockets; using Esiur.Protocol; +using Esiur.Resource; using Esiur.Security.Authority.Providers; using Esiur.Security.Cryptography; +using Esiur.Stores; using System.Net; [Collection("Integration")] public class SessionHeadersIntegrationTests { + [Fact] + public async Task AcceptedConnection_RaisesReadyAndDisconnectedLifecycleEvents() + { + var ready = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + var disconnected = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + + await using var cluster = await IntegrationCluster + .StartAsync( + _ => Task.CompletedTask, + serverCreated: server => + { + server.ConnectionReady += connection => ready.TrySetResult(connection); + server.ConnectionDisconnected += connection => disconnected.TrySetResult(connection); + }) + .WaitAsync(TimeSpan.FromSeconds(10)); + + var accepted = await ready.Task.WaitAsync(TimeSpan.FromSeconds(5)); + Assert.True(accepted.Session.Authenticated); + Assert.Equal("tester", accepted.Session.RemoteIdentity); + + accepted.NetworkClose(accepted.Socket); + Assert.Same( + accepted, + await disconnected.Task.WaitAsync(TimeSpan.FromSeconds(5))); + } + + [Fact] + public async Task AcceptedConnection_CanFetchInitiatorResourceBidirectionally() + { + var fetched = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + + await using var cluster = await IntegrationCluster + .StartAsync( + _ => Task.CompletedTask, + serverCreated: server => server.ConnectionReady += connection => + _ = Task.Run(async () => + { + try + { + fetched.TrySetResult(await connection.Get("client/exported")); + } + catch (Exception exception) + { + fetched.TrySetException(exception); + } + }), + populateClient: async warehouse => + { + await warehouse.Put("client", new MemoryStore()); + await warehouse.Put("client/exported", new Node { Id = 818 }); + }) + .WaitAsync(TimeSpan.FromSeconds(10)); + + var proxy = Assert.IsType( + await fetched.Task.WaitAsync(TimeSpan.FromSeconds(5))); + dynamic exported = proxy; + Assert.Equal(818, Convert.ToInt32(exported.Id)); + } + [Fact] public async Task AuthenticatedHandshake_StoresTypedHeadersWithoutAuthenticationData() { @@ -234,6 +298,36 @@ public class SessionHeadersIntegrationTests Assert.Equal(203, Convert.ToInt32(result)); } + [Fact] + public async Task AnonymousWebSocketTransport_AssignsSchemaDomain() + { + await using var cluster = await IntegrationCluster + .StartAsync( + async warehouse => + { + await warehouse.Put("sys/websocket-anonymous", new EncryptedEchoResource()); + }, + useWebSocket: true, + anonymous: true) + .WaitAsync(TimeSpan.FromSeconds(10)); + + var remote = (EpResource)await Task.Run(async () => + await cluster.Connection.Get("sys/websocket-anonymous")) + .WaitAsync(TimeSpan.FromSeconds(10)); + var function = remote.Instance.Definition + .GetFunctionDefByName(nameof(EncryptedEchoResource.Echo)); + var result = await remote._Invoke( + function.Index, + new Map + { + [0] = 42, + }); + + Assert.Equal(42, Convert.ToInt32(result)); + Assert.False(string.IsNullOrWhiteSpace( + Assert.Single(cluster.Server.Connections).RemoteDomain)); + } + [Fact] public async Task WrongAuthenticatedSessionKey_FailsPromptlyDuringKeyConfirmation() { diff --git a/Tests/Unit/RuntimeCasterRecordTests.cs b/Tests/Unit/RuntimeCasterRecordTests.cs new file mode 100644 index 0000000..558c6b0 --- /dev/null +++ b/Tests/Unit/RuntimeCasterRecordTests.cs @@ -0,0 +1,44 @@ +using Esiur.Data; +using Esiur.Resource; + +namespace Esiur.Tests.Unit; + +public sealed class RuntimeCasterRecordTests +{ + [Fact] + public void DynamicRecordMaterializesAsDeclaredRecordType() + { + var warehouse = new Warehouse(); + var nestedDefinition = warehouse.GetLocalTypeDefByType(typeof(NestedRecord)); + var targetDefinition = warehouse.GetLocalTypeDefByType(typeof(TargetRecord)); + var nested = new Esiur.Data.Record(nestedDefinition) + { + [nameof(NestedRecord.Name)] = "edge-1", + }; + var source = new Esiur.Data.Record(targetDefinition) + { + [nameof(TargetRecord.Enabled)] = true, + [nameof(TargetRecord.Count)] = (short)7, + [nameof(TargetRecord.Nested)] = nested, + }; + + var converted = Assert.IsType( + RuntimeCaster.Cast(source, typeof(TargetRecord))); + + Assert.True(converted.Enabled); + Assert.Equal(7, converted.Count); + Assert.Equal("edge-1", converted.Nested.Name); + } + + private sealed class TargetRecord : IRecord + { + public bool Enabled { get; set; } + public int Count { get; set; } + public NestedRecord Nested { get; set; } = new(); + } + + private sealed class NestedRecord : IRecord + { + public string Name { get; set; } = string.Empty; + } +} diff --git a/Tools/Esiur.CLI/Configuration/ConfigurationResolver.cs b/Tools/Esiur.CLI/Configuration/ConfigurationResolver.cs index c5c2e1a..9cee896 100644 --- a/Tools/Esiur.CLI/Configuration/ConfigurationResolver.cs +++ b/Tools/Esiur.CLI/Configuration/ConfigurationResolver.cs @@ -63,8 +63,8 @@ public static class ConfigurationResolver static readonly string[] ValidSchemes = ["ep", "eps", "ws", "wss"]; /// - /// Accepts ep(s)://host:port (a bare Esiur endpoint, connects at the - /// WebSocket root) as well as ws(s)://host:port/path (for hosts like + /// Accepts ep(s)://host[:port] (a bare Esiur endpoint, connects at the + /// WebSocket root) as well as ws(s)://host[:port]/path (for hosts like /// ASP.NET Core's MapEsiur("/esiur") that mount the WebSocket route /// somewhere other than root) — see for how /// the two forms are dialed. @@ -75,37 +75,12 @@ public static class ConfigurationResolver || !ValidSchemes.Contains(uri.Scheme, StringComparer.OrdinalIgnoreCase) || string.IsNullOrWhiteSpace(uri.Host) || !string.IsNullOrEmpty(uri.UserInfo) - || !HasExplicitPort(endpoint)) + || uri.Port == 0 + || uri.Port > ushort.MaxValue) throw new CliException( - $"Endpoint \"{endpoint}\" is not a valid ep:// or ws:// endpoint with an explicit port.", + $"Endpoint \"{endpoint}\" is not a valid ep:// or ws:// endpoint.", ExitCodes.InvalidArguments); } - - static bool HasExplicitPort(string endpoint) - { - var schemeEnd = endpoint.IndexOf("://", StringComparison.Ordinal); - if (schemeEnd < 0) - return false; - - var authorityStart = schemeEnd + 3; - var authorityEnd = endpoint.IndexOfAny(['/', '?', '#'], authorityStart); - if (authorityEnd < 0) - authorityEnd = endpoint.Length; - - var authority = endpoint[authorityStart..authorityEnd]; - var at = authority.LastIndexOf('@'); - if (at >= 0) - authority = authority[(at + 1)..]; - - var colon = authority.StartsWith('[') - ? authority.IndexOf(']') + 1 - : authority.LastIndexOf(':'); - return colon > 0 - && colon < authority.Length - 1 - && authority[colon] == ':' - && ushort.TryParse(authority[(colon + 1)..], out var port) - && port > 0; - } } public static class DurationParser diff --git a/Tools/Esiur.CLI/README.md b/Tools/Esiur.CLI/README.md index 9099320..b8e3da6 100644 --- a/Tools/Esiur.CLI/README.md +++ b/Tools/Esiur.CLI/README.md @@ -19,13 +19,13 @@ Runtime-specific publishes are configured as self-contained, single-file, and no Create and verify a password-authenticated profile: ```console -esiur login production "ep://host:${ESIUR_PORT}" --provider password --identity ahmed +esiur login production "ep://host" --provider password --identity ahmed ``` The password prompt does not echo input. For automation, supply it on standard input: ```console -printf '%s' "$ESIUR_PASSWORD" | esiur login production "ep://host:${ESIUR_PORT}" \ +printf '%s' "$ESIUR_PASSWORD" | esiur login production "ep://host" \ --provider password --identity ahmed --password-stdin ``` @@ -86,7 +86,7 @@ Every operational command accepts a saved profile or a temporary endpoint: ```console esiur --profile production describe sys/service -esiur --endpoint "ep://host:${ESIUR_PORT}" query sys --output json +esiur --endpoint "ep://host" query sys --output json esiur get sys/service Name --timeout 30s ```