Fix live reading examples returning data minutes behind live (ADO 91065)

docs/server-listener.md

@@ -120,7 +120,7 @@ This queries the database, loads recent sessions, and monitors for EndTime growt
 | Client can't connect | Firewall blocking the port | Open the Server Listener port (TCP+UDP) on both machines |
 | `FindLiveSession` says "no live sessions" | No active recording, or data hasn't flushed yet | Start a recorder on the ADS; use `--wait 90` for slow flush intervals |
 | `CreateParameterDataAccess` throws | Session loaded before parameters were flushed | Wait for the first data flush (~60–70s), then reload the session |
-| Data values don't change between polls | Cached session data from initial load | Reload the session (`SessionManager.Load()`) each poll cycle — see `poll-live-samples.cs` |
+| Data values don't change between polls, or arrive ~5 minutes behind the session end | Session loaded from the database instead of the Server Listener (state **LiveNotInServer**) | Enable the Server Listener on the recorder and load the session with the connection string from `SessionSummary.GetConnectionString()`; check `State == Live` first. See `poll-live-samples.cs` |
 
 ## Related
 

notebooks/06-live-data-monitoring.ipynb

@@ -75,6 +75,16 @@
    "metadata": {},
    "outputs": [],
    "source": [
+    "# CreateParameterDataAccess can return None (e.g. for a composite session); skip those parameters.\n",
+    "pdas = {}\n",
+    "for param_id in monitor_params:\n",
+    "    pda = session.CreateParameterDataAccess(param_id)\n",
+    "    if pda is None:\n",
+    "        print(f\"Skipping {param_id}: could not create a ParameterDataAccess.\")\n",
+    "    else:\n",
+    "        pdas[param_id] = pda\n",
+    "monitor_params = list(pdas)\n",
+    "\n",
     "history = {p: [] for p in monitor_params}\n",
     "timestamps = []\n",
     "\n",
@@ -90,18 +100,14 @@
     "        start_time = max(session.StartTime, end_time - window_ns)\n",
     "\n",
     "        latest = {}\n",
-    "        for param_id in monitor_params:\n",
-    "            pda = session.CreateParameterDataAccess(param_id)\n",
-    "            try:\n",
-    "                samples = pda.GetSamplesBetween(start_time, end_time)\n",
-    "                if samples.SampleCount > 0:\n",
-    "                    val = float(samples.Data(samples.SampleCount - 1))\n",
-    "                    latest[param_id] = val\n",
-    "                    history[param_id].append(val)\n",
-    "                else:\n",
-    "                    latest[param_id] = float('nan')\n",
-    "            finally:\n",
-    "                pda.Dispose()\n",
+    "        for param_id, pda in pdas.items():\n",
+    "            samples = pda.GetSamplesBetween(start_time, end_time)\n",
+    "            if samples.SampleCount > 0:\n",
+    "                val = float(samples.Data[samples.SampleCount - 1])\n",
+    "                latest[param_id] = val\n",
+    "                history[param_id].append(val)\n",
+    "            else:\n",
+    "                latest[param_id] = float('nan')\n",
     "\n",
     "        timestamps.append(poll * POLL_INTERVAL)\n",
     "\n",
@@ -117,7 +123,9 @@
     "except KeyboardInterrupt:\n",
     "    print(\"\\nPolling stopped by user.\")\n",
     "\n",
-    "print(f\"\\nCollected {len(timestamps)} samples.\")"
+    "print(f\"\\nCollected {len(timestamps)} samples.\")\n",
+    "for pda in pdas.values():\n",
+    "    pda.Dispose()"
    ]
   },
   {

notebooks/sqlrace_helpers.py

@@ -111,7 +111,18 @@ def load_session(session_manager, session_guid, connection_string=None):
 
     key = SessionKey.Parse(session_guid)
     cs = connection_string or get_connection_string()
-    client_session = session_manager.Load(key, cs)
+    summary = session_manager.FindSummaryBy(key, cs)
+    if summary is None:
+        raise LookupError(f"Session {session_guid} not found in {cs}")
+
+    # For a Live session this is the recorder's Server Listener connection string.
+    # A LiveNotInServer session has no listener, so its database copy trails the
+    # leading edge by the recorder's flush delay (5 minutes by default).
+    if str(summary.State) == "LiveNotInServer":
+        print("WARNING: Session is LiveNotInServer - enable the recorder's Server Listener "
+              "for live data (docs/server-listener.md).")
+
+    client_session = session_manager.Load(summary.Key, summary.GetConnectionString())
     return client_session, client_session.Session
 
 

projects/csharp/LiveDashboard/Program.cs

@@ -1,9 +1,12 @@
+using LiveDashboard;
+
 using MESL.SqlRace.Domain;
-using MESL.SqlRace.Domain.Infrastructure.Enumerators;
+using MESL.SqlRace.Domain.Infrastructure.DataPipeline;
 using MESL.SqlRace.Domain.Query;
+
 using Microsoft.Extensions.Configuration;
+
 using SqlRace.Examples.Core;
-using LiveDashboard;
 
 // ─── Configuration ───────────────────────────────────────────
 var configuration = new ConfigurationBuilder()
@@ -48,7 +51,12 @@
 var sessionManager = SessionManager.CreateSessionManager();
 IClientSession? clientSession = null;
 Session? session = null;
-dynamic? loadKey = null; // Store the key for reloading
+
+// A LiveNotInServer session is still recording but has no Server Listener, so the only
+// copy is the database's, which trails the leading edge by the recorder's flush delay
+// (5 minutes by default). Its EndTime is also fixed at load, so the dashboard would freeze.
+const string NoListenerHelp =
+    "Enable the Server Listener on the recorder for live data - see docs/server-listener.md.";
 
 if (!string.IsNullOrEmpty(configuredSessionId))
 {
@@ -57,54 +65,74 @@
     {
         Console.WriteLine($"Session '{configuredSessionId}' not found. Available:");
         foreach (var s in sessions.OrderByDescending(s => s.TimeOfRecording).Take(5))
+        {
             Console.WriteLine($"  {s.Identifier} ({s.Key})");
+        }
+
         return;
     }
 
-    clientSession = sessionManager.Load(match.Key, connectionString);
+    if (match.State == SessionState.LiveNotInServer)
+    {
+        Console.WriteLine($"Session '{configuredSessionId}' is LiveNotInServer. {NoListenerHelp}");
+        return;
+    }
+
+    var sessionConnectionString = match.GetConnectionString();
+    clientSession = sessionManager.Load(match.Key, sessionConnectionString);
     session = clientSession.Session;
-    loadKey = match.Key;
     Console.WriteLine($"Loading configured session: {match.Identifier} ({match.Key})");
 }
 else
 {
     // Try sessions newest-first, skip any with no sample data
     foreach (var candidate in sessions.OrderByDescending(s => s.TimeOfRecording))
     {
-        var tempClient = sessionManager.Load(candidate.Key, connectionString);
+        if (candidate.State == SessionState.LiveNotInServer)
+        {
+            Console.WriteLine($"Skipping session {candidate.Identifier} (LiveNotInServer). {NoListenerHelp}");
+            continue;
+        }
+
+        var sessionConnectionString = candidate.GetConnectionString();
+        var tempClient = sessionManager.Load(candidate.Key, sessionConnectionString);
         var tempSession = tempClient.Session;
 
         // Check if the first configured parameter has any data
         var testParam = paramIds.FirstOrDefault(id =>
             tempSession.Parameters.Any(p => p.Identifier == id));
 
-        if (testParam != null)
+        if (testParam == null)
         {
-            using var testPda = tempSession.CreateParameterDataAccess(testParam);
-            testPda.GoTo(long.MaxValue);
-            var testSamples = testPda.GetNextSamples(1, StepDirection.Reverse);
-
-            if (testSamples.SampleCount > 0)
-            {
-                clientSession = tempClient;
-                session = tempSession;
-                loadKey = candidate.Key;
-                Console.WriteLine($"Loading session: {candidate.Identifier} ({candidate.Key})");
-                break;
-            }
+            tempClient.Close();
+            Console.WriteLine($"Skipping session {candidate.Identifier} (test parameter not found)");
+            continue;
         }
 
-        tempClient.Dispose();
-        Console.WriteLine($"Skipping session {candidate.Identifier} (no sample data)");
-    }
+        using var testPda = tempSession.CreateParameterDataAccess(testParam);
+        var sampleCount = testPda.GetSamplesCount(tempSession.StartTime, tempSession.EndTime);
 
-    if (session == null)
-    {
-        Console.WriteLine("No sessions with sample data found.");
-        return;
+        if (sampleCount == 0)
+        {
+            Console.WriteLine($"Skipping session {candidate.Identifier} (no sample data)");
+            tempClient.Close();
+            continue;
+        }
+
+        clientSession = tempClient;
+        session = tempSession;
+        Console.WriteLine($"Loading session: {candidate.Identifier} ({candidate.Key})");
+        break;
     }
 }
 
+if (clientSession == null ||
+    session == null)
+{
+    Console.WriteLine("No sessions with sample data found.");
+    return;
+}
+
 // ─── Resolve available parameters ────────────────────────────
 var availableParams = paramIds
     .Where(id => session.Parameters.Any(p => p.Identifier == id))
@@ -121,61 +149,98 @@
 var renderer = new DashboardRenderer(pollingMs);
 var sessionId = session.Identifier;
 
+var lastEndTime = long.MinValue;
+// Create a list of PDAs for the available parameters.
+var parameterPdas = new List<ParameterDataAccessBase>(availableParams.Count);
+foreach (var param in availableParams)
+{
+    var pda = session.CreateParameterDataAccess(param);
+    parameterPdas.Add(pda);
+}
+
 while (!cts.Token.IsCancellationRequested)
 {
-    // Reload session each cycle to pick up newly flushed data
-    clientSession?.Dispose();
-    clientSession = sessionManager.Load(loadKey!, connectionString);
-    session = clientSession.Session;
+    try
+    {
+        await Task.Delay(pollingMs, cts.Token);
+    }
+    catch (OperationCanceledException)
+    {
+        break;
+    }
 
     var stats = new List<DashboardRenderer.ParameterStats>();
-
-    foreach (var paramId in availableParams)
+    if (lastEndTime >= session.EndTime)
     {
-        // Create a fresh PDA each cycle so we always see the latest live data
-        using var pda = session.CreateParameterDataAccess(paramId);
-        pda.GoTo(long.MaxValue);
-        var samples = pda.GetNextSamples(windowSize, StepDirection.Reverse);
+        // -- No updates to the session, skip this polling cycle --
+        continue;
+    }
 
-        if (samples.SampleCount == 0)
+    lastEndTime = session.EndTime;
+
+    foreach (var pda in parameterPdas)
+    {
+        // Query the PDA for new samples
+        var pdaEndTime = pda.GetEndTime();
+        if (pdaEndTime == null)
         {
-            stats.Add(new DashboardRenderer.ParameterStats(
-                paramId, double.NaN, double.NaN, double.NaN, double.NaN, 0));
+            // No samples are available for the PDA.
+            stats.Add(
+                new DashboardRenderer.ParameterStats(
+                    pda.ParameterIdentifier,
+                    double.NaN,
+                    double.NaN,
+                    double.NaN,
+                    double.NaN,
+                    0));
             continue;
         }
 
-        var latest = samples.Data[0];
-        var min = double.MaxValue;
-        var max = double.MinValue;
-        var sum = 0.0;
+        var maxSampleIntervalNs = (long)(1_000_000_000 / pda.GetMaximumFrequency());
+        var duration = maxSampleIntervalNs * windowSize;
+        var startTime = pdaEndTime.Value - duration;
+        // Grab specifically the end, Max, Min, and Average values for the parameter over the window.
+        var dataStatistics = pda.GetDataStatistics(
+            startTime,
+            duration,
+            statisticOption: StatisticOption.End | StatisticOption.Max | StatisticOption.Min | StatisticOption.Mean);
 
-        for (var j = 0; j < samples.SampleCount; j++)
+        if (dataStatistics.NumberOfSamples == 0)
         {
-            var v = samples.Data[j];
-            if (v < min) min = v;
-            if (v > max) max = v;
-            sum += v;
+            // No samples are available for the PDA in the specified window.
+            stats.Add(
+                new DashboardRenderer.ParameterStats(
+                    pda.ParameterIdentifier,
+                    double.NaN,
+                    double.NaN,
+                    double.NaN,
+                    double.NaN,
+                    0));
+            continue;
         }
 
-        stats.Add(new DashboardRenderer.ParameterStats(
-            paramId, latest, min, max, sum / samples.SampleCount, samples.SampleCount));
+        var latest = dataStatistics.EndValue;
+        var min = dataStatistics.MinimumValue;
+        var max = dataStatistics.MaximumValue;
+
+        stats.Add(
+            new DashboardRenderer.ParameterStats(
+                pda.ParameterIdentifier,
+                latest,
+                min,
+                max,
+                dataStatistics.MeanValue,
+                dataStatistics.NumberOfSamples));
     }
 
     renderer.Render(
         sessionId,
         session.EndTime > session.StartTime ? "Data" : "Empty",
         stats);
-
-    try
-    {
-        await Task.Delay(pollingMs, cts.Token);
-    }
-    catch (OperationCanceledException)
-    {
-        break;
-    }
 }
 
 Console.SetCursorPosition(0, Console.CursorTop + 1);
 Console.WriteLine("Dashboard stopped.");
-clientSession?.Dispose();
+// Dispose all PDAs once we are done.
+parameterPdas.ForEach(x => x.Dispose());
+clientSession.Close();