GVWIE

experiments/CrossLanguageRecoveryServer/CrossLanguageRecoveryServer.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net10.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\Libraries\Esiur\Esiur.csproj" OutputItemType="Analyzer" />
+  </ItemGroup>
+
+</Project>

experiments/CrossLanguageRecoveryServer/Program.cs

@@ -0,0 +1,108 @@
+using System.Text.Json;
+using Esiur.Protocol;
+using Esiur.Resource;
+using Esiur.Stores;
+using Esiur.Experiments.CrossLanguageRecoveryServer;
+
+var host = GetArg(args, "--host", "127.0.0.1");
+var port = int.Parse(GetArg(args, "--port", "10901"));
+var updatePeriodMs = int.Parse(GetArg(args, "--update-period", "100"));
+var outputDirectory = Path.GetFullPath(GetArg(args, "--output", Path.Combine("results", "cross-language-recovery")));
+var waitForStdin = !HasFlag(args, "--no-stdin");
+
+Directory.CreateDirectory(outputDirectory);
+
+var shutdown = new CancellationTokenSource();
+Console.CancelKeyPress += (_, e) =>
+{
+    e.Cancel = true;
+    shutdown.Cancel();
+};
+
+var wh = new Warehouse();
+await wh.Put("sys", new MemoryStore());
+var epServer = new EpServer
+{
+    Port = (ushort)port,
+    AllowUnauthorizedAccess = true
+};
+
+if (!string.IsNullOrWhiteSpace(host) && host != "0.0.0.0")
+    epServer.IP = host;
+
+var server = await wh.Put("sys/server", epServer);
+
+var resource = await wh.Put("sys/recovery", new RecoveryTestResource(outputDirectory));
+await wh.Open();
+resource.SetStatus("ready");
+resource.StartPeriodicUpdates(updatePeriodMs, shutdown.Token);
+resource.AppendLog("server_started");
+
+var ready = new
+{
+    host,
+    port,
+    url = $"ep://{host}:{port}",
+    websocket_url = $"ws://{host}:{port}",
+    resource_path = "sys/recovery",
+    update_period_ms = updatePeriodMs,
+    output_directory = outputDirectory,
+    started_utc = DateTimeOffset.UtcNow
+};
+
+var readyPath = Path.Combine(outputDirectory, "server-ready.json");
+await File.WriteAllTextAsync(readyPath, JsonSerializer.Serialize(ready, new JsonSerializerOptions { WriteIndented = true }));
+
+Console.WriteLine($"[CrossLanguageRecoveryServer] listening ep://{host}:{port}/sys/recovery");
+Console.WriteLine($"[CrossLanguageRecoveryServer] updatePeriodMs={updatePeriodMs}");
+Console.WriteLine($"[CrossLanguageRecoveryServer] output={outputDirectory}");
+Console.WriteLine(waitForStdin
+    ? "[CrossLanguageRecoveryServer] Press ENTER or Ctrl+C to stop."
+    : "[CrossLanguageRecoveryServer] Running until the process is stopped.");
+
+_ = Task.Run(async () =>
+{
+    try
+    {
+        while (!shutdown.IsCancellationRequested)
+        {
+            await Task.Delay(1000, shutdown.Token).ConfigureAwait(false);
+            var state = resource.CreateSnapshot();
+            Console.WriteLine($"[CrossLanguageRecoveryServer] counter={state.Counter} status={state.Status} age={state.Age} clients={server.Connections.Count}");
+        }
+    }
+    catch (TaskCanceledException)
+    {
+    }
+});
+
+if (waitForStdin)
+{
+    _ = Task.Run(() =>
+    {
+        Console.ReadLine();
+        shutdown.Cancel();
+    });
+}
+
+try
+{
+    await Task.Delay(Timeout.InfiniteTimeSpan, shutdown.Token);
+}
+catch (TaskCanceledException)
+{
+}
+finally
+{
+    resource.AppendLog("server_stopping");
+    resource.StopPeriodicUpdates();
+    await wh.Close();
+}
+
+static string GetArg(string[] args, string key, string defaultValue)
+{
+    var i = Array.IndexOf(args, key);
+    return i >= 0 && i + 1 < args.Length ? args[i + 1] : defaultValue;
+}
+
+static bool HasFlag(string[] args, string key) => args.Contains(key);

experiments/CrossLanguageRecoveryServer/RecoveryTestResource.cs

@@ -0,0 +1,128 @@
+using System.Text.Json;
+using Esiur.Resource;
+
+namespace Esiur.Experiments.CrossLanguageRecoveryServer;
+
+[Resource]
+[Annotation("experiment", "Cross-Language TypeDef Discovery and Reattachment Recovery")]
+public partial class RecoveryTestResource
+{
+    readonly object sync = new();
+    readonly string logPath;
+    CancellationTokenSource? updatesCts;
+    volatile bool updatesPaused;
+
+    [Export]
+    [Annotation("semantic", "monotonic counter incremented by the C# server")]
+    int counter;
+
+    [Export]
+    [Annotation("semantic", "client-visible service status")]
+    string status = "starting";
+
+    [Export]
+    [Annotation("semantic", "UTC tick timestamp of the last counter update")]
+    long lastUpdateTicks;
+
+    [Export]
+    [Annotation("semantic", "raised whenever Counter is incremented")]
+    public event ResourceEventHandler<int>? CounterChanged;
+
+    public RecoveryTestResource(string outputDirectory)
+    {
+        Directory.CreateDirectory(outputDirectory);
+        logPath = Path.Combine(outputDirectory, "server_log.jsonl");
+        File.WriteAllText(logPath, "");
+    }
+
+    [Export]
+    [Annotation("semantic", "returns a + b; used to verify dynamic function invocation")]
+    public int Add(int a, int b) => a + b;
+
+    [Export]
+    [Annotation("semantic", "sets Status and returns true")]
+    public bool SetStatus(string value)
+    {
+        Status = value;
+        return true;
+    }
+
+    [Export]
+    [Annotation("semantic", "pauses or resumes periodic server updates")]
+    public bool SetUpdatesPaused(bool paused)
+    {
+        updatesPaused = paused;
+        return true;
+    }
+
+    [Export]
+    [Annotation("semantic", "authoritative C# state snapshot encoded as JSON")]
+    public string GetAuthoritativeStateJson() => JsonSerializer.Serialize(CreateSnapshot());
+
+    public void StartPeriodicUpdates(int updatePeriodMs, CancellationToken shutdown)
+    {
+        updatesCts = CancellationTokenSource.CreateLinkedTokenSource(shutdown);
+        var token = updatesCts.Token;
+        _ = Task.Run(async () =>
+        {
+            while (!token.IsCancellationRequested)
+            {
+                await Task.Delay(updatePeriodMs, token).ConfigureAwait(false);
+                if (updatesPaused)
+                    continue;
+
+                int value;
+                lock (sync)
+                {
+                    Counter = Counter + 1;
+                    LastUpdateTicks = DateTime.UtcNow.Ticks;
+                    value = Counter;
+                }
+
+                CounterChanged?.Invoke(value);
+                AppendLog("tick");
+            }
+        }, token);
+    }
+
+    public void StopPeriodicUpdates()
+    {
+        updatesCts?.Cancel();
+    }
+
+    public RecoveryStateSnapshot CreateSnapshot()
+    {
+        lock (sync)
+        {
+            return new RecoveryStateSnapshot
+            {
+                Counter = Counter,
+                Status = Status,
+                LastUpdateTicks = LastUpdateTicks,
+                Age = Instance?.Age ?? 0,
+                TimestampUtc = DateTimeOffset.UtcNow,
+                UpdatesPaused = updatesPaused
+            };
+        }
+    }
+
+    public void AppendLog(string eventName)
+    {
+        var payload = new
+        {
+            event_name = eventName,
+            state = CreateSnapshot()
+        };
+        File.AppendAllText(logPath, JsonSerializer.Serialize(payload) + Environment.NewLine);
+    }
+}
+
+public sealed class RecoveryStateSnapshot
+{
+    public int Counter { get; set; }
+    public string Status { get; set; } = "";
+    public long LastUpdateTicks { get; set; }
+    public ulong Age { get; set; }
+    public DateTimeOffset TimestampUtc { get; set; }
+    public bool UpdatesPaused { get; set; }
+}