using packet timestamp

Client/WebUI/hub.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"encoding/binary"
 	"encoding/json"
 	"log"
 	"math"
@@ -15,10 +16,15 @@ import (
 
 // ─── WebSocket client ─────────────────────────────────────────────────────────
 
+type wsMessage struct {
+	msgType int
+	data    []byte
+}
+
 type wsClient struct {
 	hub  *Hub
 	conn *websocket.Conn
-	send chan []byte
+	send chan wsMessage
 }
 
 func (c *wsClient) writePump() {
@@ -34,7 +40,7 @@ func (c *wsClient) writePump() {
 				c.conn.WriteMessage(websocket.CloseMessage, []byte{})
 				return
 			}
-			if err := c.conn.WriteMessage(websocket.TextMessage, msg); err != nil {
+			if err := c.conn.WriteMessage(msg.msgType, msg.data); err != nil {
 				return
 			}
 		case <-pingTicker.C:
@@ -69,7 +75,7 @@ func (c *wsClient) readPump() {
 				case "ping":
 					resp, _ := json.Marshal(map[string]string{"type": "pong"})
 					select {
-					case c.send <- resp:
+					case c.send <- wsMessage{websocket.TextMessage, resp}:
 					default:
 					}
 				case "addSource":
@@ -167,8 +173,8 @@ func NewHub() *Hub {
 		clients:     make(map[*wsClient]bool),
 		register:    make(chan *wsClient, 8),
 		unregister:  make(chan *wsClient, 8),
-		broadcastCh: make(chan []byte, 64),
-		dataCh:      make(chan taggedSample, 256),
+		broadcastCh: make(chan []byte, 256),
+		dataCh:      make(chan taggedSample, 65536), // large buffer: absorbs bursts at high sample rates
 		commandCh:   make(chan hubCmd, 64),
 		rings:       make(map[string]*sigRing),
 		statsMap:    make(map[string]*SourceStat),
@@ -298,7 +304,7 @@ func (h *Hub) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
 		log.Printf("ws upgrade: %v", err)
 		return
 	}
-	c := &wsClient{hub: h, conn: conn, send: make(chan []byte, 64)}
+	c := &wsClient{hub: h, conn: conn, send: make(chan wsMessage, 64)}
 	h.register <- c
 	go c.writePump()
 	go c.readPump()
@@ -345,11 +351,11 @@ func (h *Hub) Run() {
 			h.clients[c] = true
 			// Send current state to the new client.
 			if sourcesMsg != nil {
-				select { case c.send <- sourcesMsg: default: }
+				select { case c.send <- wsMessage{websocket.TextMessage, sourcesMsg}: default: }
 			}
 			for _, src := range sourcesMap {
 				if src.configJS != nil {
-					select { case c.send <- src.configJS: default: }
+					select { case c.send <- wsMessage{websocket.TextMessage, src.configJS}: default: }
 				}
 			}
 
@@ -361,7 +367,7 @@ func (h *Hub) Run() {
 
 		case msg := <-h.broadcastCh:
 			for c := range h.clients {
-				select { case c.send <- msg: default: }
+				select { case c.send <- wsMessage{websocket.TextMessage, msg}: default: }
 			}
 
 		case cmd := <-h.commandCh:
@@ -473,10 +479,15 @@ func (h *Hub) Run() {
 					pending[srcID] = pending[srcID][:0]
 					continue
 				}
-				msg := h.buildDataMessageForSource(src, samples)
+				msg := h.buildBinaryDataMessageForSource(src, samples)
 				pending[srcID] = pending[srcID][:0]
 				if msg != nil {
-					h.broadcast(msg)
+					for c := range h.clients {
+						select {
+						case c.send <- wsMessage{websocket.BinaryMessage, msg}:
+						default:
+						}
+					}
 				}
 			}
 
@@ -763,6 +774,223 @@ func (h *Hub) buildDataMessageForSource(src *sourceHubState, batch []DataSample)
 	return result
 }
 
+// buildBinaryDataMessageForSource encodes a batch of samples as a compact binary
+// frame for WebSocket binary messages. Skips the JSON overhead entirely.
+//
+// Wire format (little-endian):
+//
+//	uint8  version (1)
+//	uint8  source ID length
+//	UTF-8  source ID
+//	uint32 number of signals
+//	for each signal:
+//	  uint16 key length
+//	  UTF-8  key (relative to source, e.g. "sigName" not "s1:sigName")
+//	  uint32 pair count N
+//	  float64[N]  t values
+//	  float64[N]  v values
+func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []DataSample) []byte {
+	if len(batch) == 0 {
+		return nil
+	}
+	if src.configSeq != src.configSeqAtCalib {
+		src.configSeqAtCalib = src.configSeq
+		src.timeSigCalib = make(map[string]float64)
+	}
+
+	sigs := src.signals
+	pfx := src.id + ":"
+
+	// ---- Phase 1: collect (t,v) for each signal (same logic as JSON path) ----
+
+	type pairBuf struct {
+		t, v []float64
+	}
+	pairs := make(map[string]pairBuf, len(sigs)*2)
+
+	for _, sig := range sigs {
+		n := sig.NumElements()
+
+		switch {
+		case n > 1 && (sig.TimeMode == TimeModeFirstSample || sig.TimeMode == TimeModeLastSample):
+			hasTimeSig := sig.TimeSignalIdx != NoTimeSignal && int(sig.TimeSignalIdx) < len(sigs)
+			var timeSigName string
+			timerToSec := 1e-6
+			if hasTimeSig {
+				ts := sigs[sig.TimeSignalIdx]
+				timeSigName = ts.Name
+				if ts.TypeCode == 6 {
+					timerToSec = 1e-9
+				}
+			}
+			dt := 0.0
+			if sig.SamplingRate > 0 {
+				dt = 1.0 / sig.SamplingRate
+			}
+			allT := make([]float64, 0, len(batch)*n)
+			allV := make([]float64, 0, len(batch)*n)
+			for _, s := range batch {
+				vals, ok := s.Values[sig.Name]
+				if !ok || len(vals) < n {
+					continue
+				}
+				var anchorTime float64
+				anchorIsFirstSample := sig.TimeMode == TimeModeFirstSample
+				if hasTimeSig {
+					tVals, tOk := s.Values[timeSigName]
+					if tOk && len(tVals) >= 1 {
+						timerS := tVals[0] * timerToSec
+						wallT := float64(s.WallTime.UnixNano()) / 1e9
+						if _, exists := src.timeSigCalib[timeSigName]; !exists {
+							src.timeSigCalib[timeSigName] = wallT - timerS
+						}
+						anchorTime = src.timeSigCalib[timeSigName] + timerS
+					} else {
+						anchorTime = float64(s.WallTime.UnixNano()) / 1e9
+						anchorIsFirstSample = false
+					}
+				} else {
+					anchorTime = float64(s.WallTime.UnixNano()) / 1e9
+					anchorIsFirstSample = false
+				}
+				for k := 0; k < n; k++ {
+					var t float64
+					if anchorIsFirstSample {
+						t = anchorTime + float64(k)*dt
+					} else {
+						t = anchorTime - float64(n-1-k)*dt
+					}
+					allT = append(allT, t)
+					allV = append(allV, vals[k])
+				}
+			}
+			// Write hi-res LTTB data to ring.
+			ringT, ringV := lttbDecimate(allT, allV, maxRingPoints)
+			if rb := h.getRing(pfx + sig.Name); rb != nil {
+				rb.write(ringT, ringV)
+			}
+			// Decimate for push.
+			decimT, decimV := lttbDecimate(allT, allV, maxPushPoints)
+			pairs[sig.Name] = pairBuf{t: decimT, v: decimV}
+
+		case n > 1 && sig.TimeMode == TimeModeFullArray:
+			hasTimeSig := sig.TimeSignalIdx != NoTimeSignal && int(sig.TimeSignalIdx) < len(sigs)
+			var timeSigName string
+			timerToSec := 1e-6
+			if hasTimeSig {
+				ts := sigs[sig.TimeSignalIdx]
+				timeSigName = ts.Name
+				if ts.TypeCode == 6 {
+					timerToSec = 1e-9
+				}
+			}
+			allT := make([]float64, 0, len(batch)*n)
+			allV := make([]float64, 0, len(batch)*n)
+			for _, s := range batch {
+				vals, ok := s.Values[sig.Name]
+				if !ok || len(vals) < n {
+					continue
+				}
+				if hasTimeSig {
+					tVals, tOk := s.Values[timeSigName]
+					if tOk && len(tVals) >= n {
+						if _, exists := src.timeSigCalib[timeSigName]; !exists {
+							wallT := float64(s.WallTime.UnixNano()) / 1e9
+							src.timeSigCalib[timeSigName] = wallT - tVals[0]*timerToSec
+						}
+						calib := src.timeSigCalib[timeSigName]
+						for k := 0; k < n; k++ {
+							allT = append(allT, calib+tVals[k]*timerToSec)
+							allV = append(allV, vals[k])
+						}
+						continue
+					}
+				}
+				wallT := float64(s.WallTime.UnixNano()) / 1e9
+				for k := 0; k < n; k++ {
+					allT = append(allT, wallT)
+					allV = append(allV, vals[k])
+				}
+			}
+			ringT, ringV := lttbDecimate(allT, allV, maxRingPoints)
+			if rb := h.getRing(pfx + sig.Name); rb != nil {
+				rb.write(ringT, ringV)
+			}
+			decimT, decimV := lttbDecimate(allT, allV, maxPushPoints)
+			pairs[sig.Name] = pairBuf{t: decimT, v: decimV}
+
+		case n == 1:
+			ts := make([]float64, 0, len(batch))
+			vs := make([]float64, 0, len(batch))
+			for _, s := range batch {
+				vals, ok := s.Values[sig.Name]
+				if !ok || len(vals) < 1 {
+					continue
+				}
+				ts = append(ts, float64(s.WallTime.UnixNano())/1e9)
+				vs = append(vs, vals[0])
+			}
+			if rb := h.getRing(pfx + sig.Name); rb != nil {
+				rb.write(ts, vs)
+			}
+			pairs[sig.Name] = pairBuf{t: ts, v: vs}
+
+		default:
+			for i := 0; i < n; i++ {
+				key := arrayKey(sig.Name, i)
+				ts := make([]float64, 0, len(batch))
+				vs := make([]float64, 0, len(batch))
+				for _, s := range batch {
+					vals, ok := s.Values[sig.Name]
+					if !ok || len(vals) <= i {
+						continue
+					}
+					ts = append(ts, float64(s.WallTime.UnixNano())/1e9)
+					vs = append(vs, vals[i])
+				}
+				if rb := h.getRing(pfx + key); rb != nil {
+					rb.write(ts, vs)
+				}
+				pairs[key] = pairBuf{t: ts, v: vs}
+			}
+		}
+	}
+
+	// ---- Phase 2: compute total size for pre-allocation ----
+	totalSize := 1 + 1 + len(src.id) + 4 // version + srcIdLen + srcId + numSigs
+	for key, p := range pairs {
+		totalSize += 2 + len(key) + 4 // keyLen + key + pairCount
+		totalSize += len(p.t) * 16   // t + v, each float64 = 8 bytes
+	}
+
+	buf := make([]byte, totalSize)
+	buf[0] = 1 // version
+	buf[1] = byte(len(src.id))
+	copy(buf[2:], src.id)
+	off := 2 + len(src.id)
+	binary.LittleEndian.PutUint32(buf[off:], uint32(len(pairs)))
+	off += 4
+
+	for key, p := range pairs {
+		binary.LittleEndian.PutUint16(buf[off:], uint16(len(key)))
+		off += 2
+		copy(buf[off:], key)
+		off += len(key)
+		binary.LittleEndian.PutUint32(buf[off:], uint32(len(p.t)))
+		off += 4
+		for i := 0; i < len(p.t); i++ {
+			binary.LittleEndian.PutUint64(buf[off:], math.Float64bits(p.t[i]))
+			off += 8
+		}
+		for i := 0; i < len(p.v); i++ {
+			binary.LittleEndian.PutUint64(buf[off:], math.Float64bits(p.v[i]))
+			off += 8
+		}
+	}
+
+	return buf
+}
+
 // RecordDataFragment is called by UDPClient for every incoming DATA datagram.
 func (h *Hub) RecordDataFragment(sourceID string, counter uint32, nBytes int, arrivalNs int64, complete bool) {
 	h.statsMu.RLock()

Client/WebUI/static/app.js

@@ -99,6 +99,7 @@ function trigPostSec() { return trig.windowSec * (100 - trig.prePercent) / 100;
 let ws = null, wsBackoff = 1000;
 function connectWS() {
   ws = new WebSocket('ws://' + location.host + '/ws');
+  ws.binaryType = 'arraybuffer';
   ws.onopen = () => { wsBackoff = 1000; setStatus('orange', 'Connected – waiting for data'); };
   ws.onclose = () => {
     setStatus('red', 'Disconnected (reconnecting…)');
@@ -107,6 +108,7 @@ function connectWS() {
   };
   ws.onerror = () => { };
   ws.onmessage = evt => {
+    if (evt.data instanceof ArrayBuffer) { onBinaryData(evt.data); return; }
     let msg; try { msg = JSON.parse(evt.data); } catch { return; }
     if (msg.type === 'sources') onSources(msg);
     else if (msg.type === 'config') onConfig(msg);
@@ -200,6 +202,8 @@ function onData(msg) {
     const len = Math.min(sd.t.length, sd.v.length);
     for (let i = 0; i < len; i++) pushBuffer(buf, sd.t[i], sd.v[i]);
   });
+  // Increment data generation counter so render loop knows data changed
+  _dataGen++;
   if (trig.enabled && trig.armed && trig.signal) checkTrigger(sigs);
   if (trig.enabled && trig.collecting && (Date.now() / 1000) >= trig.trigTime + trigPostSec())
     finaliseTriggerCapture();
@@ -212,8 +216,81 @@ function onData(msg) {
 }
 
 /* ════════════════════════════════════════════════════════════════
-   Trigger logic
+   Binary data handler — parses compact binary frames from Go backend.
+   Wire format (little-endian):
+     uint8  version (1)
+     uint8  sourceIdLen
+     UTF-8  sourceId
+     uint32 numSignals
+     for each signal:
+       uint16 keyLen
+       UTF-8  key (relative to source)
+       uint32 pairCount N
+       float64[N] t values
+       float64[N] v values
    ════════════════════════════════════════════════════════════════ */
+function onBinaryData(buf) {
+  lastDataAt = performance.now();
+  const dv = new DataView(buf);
+  let off = 0;
+  if (dv.getUint8(off) !== 1) return;
+  off += 1;
+
+  const srcIdLen = dv.getUint8(off); off += 1;
+  const srcId = new TextDecoder().decode(new Uint8Array(buf, off, srcIdLen));
+  off += srcIdLen;
+  const prefix = srcId + ':';
+
+  const numSigs = dv.getUint32(off, true); off += 4;
+
+  // Collect trigger-signal values for inline check
+  let trigVals = null;
+
+  for (let s = 0; s < numSigs; s++) {
+    const keyLen = dv.getUint16(off, true); off += 2;
+    const key = new TextDecoder().decode(new Uint8Array(buf, off, keyLen));
+    off += keyLen;
+    const fullKey = prefix + key;
+
+    const n = dv.getUint32(off, true); off += 4;
+    let bufObj = buffers[fullKey];
+    if (!bufObj) {
+      bufObj = makeBuffer(n > 100 ? TEMPORAL_CAP : DEFAULT_CAP);
+      buffers[fullKey] = bufObj;
+    }
+
+    // Read t and v values in one pass (v array starts at off + n*8)
+    const tOff = off, vOff = off + n * 8;
+    for (let i = 0; i < n; i++) {
+      pushBuffer(bufObj, dv.getFloat64(tOff + i * 8, true), dv.getFloat64(vOff + i * 8, true));
+    }
+    off += n * 16; // skip both t and v arrays
+
+    // Capture trigger signal values
+    if (trig.enabled && trig.armed && fullKey === trig.signal) {
+      trigVals = { t: new Float64Array(n), v: new Float64Array(n) };
+      for (let i = 0; i < n; i++) {
+        trigVals.t[i] = dv.getFloat64(tOff + i * 8, true);
+        trigVals.v[i] = dv.getFloat64(vOff + i * 8, true);
+      }
+    }
+  }
+
+  // Trigger check
+  if (trigVals) checkTrigger(trigVals);
+
+  if (trig.enabled && trig.collecting && (Date.now() / 1000) >= trig.trigTime + trigPostSec())
+    finaliseTriggerCapture();
+
+  if (!trig.enabled) {
+    _dataGen++;
+    plots.forEach(p => {
+      if (globalPause) return;
+      if (p.traces.some(t => buffers[t] !== undefined)) p.needsRedraw = true;
+    });
+  }
+}
+
 function checkTrigger(sigs) {
   const sd = sigs[trig.signal]; if (!sd || !sd.v || !sd.v.length) return;
   for (let i = 0; i < sd.v.length; i++) {
@@ -747,8 +824,11 @@ function drawCursorLines(u, p) {
 }
 
 // Compute the rolling-window anchor ("newest common timestamp") for a plot.
-// Returns the min-of-max timestamp across all sources contributing traces to p,
-// so no source shows a blank right edge.  Falls back to Date.now()/1000 if no data.
+// Returns the min-of-max timestamp across ACTIVE sources contributing traces to p,
+// so no live source shows a blank right edge.
+// Sources whose newest timestamp lags the fastest source by more than windowSec are
+// considered stale (disconnected / from a previous session) and are excluded, so they
+// cannot anchor the rolling window far in the past.
 function computePlotNow(p) {
   const sourceNewest = {};
   p.traces.forEach(key => {
@@ -761,7 +841,11 @@ function computePlotNow(p) {
     if (sourceNewest[srcId] === undefined || t > sourceNewest[srcId]) sourceNewest[srcId] = t;
   });
   const srcVals = Object.values(sourceNewest);
-  let now = srcVals.length > 0 ? Math.min(...srcVals) : -Infinity;
+  if (srcVals.length === 0) return Date.now() / 1000;
+  const globalMax = Math.max(...srcVals);
+  // Keep only sources that have received data within the last windowSec.
+  const active = srcVals.filter(t => t >= globalMax - windowSec);
+  let now = active.length > 0 ? Math.min(...active) : globalMax;
   if (!isFinite(now)) now = Date.now() / 1000;
   return now;
 }
@@ -1031,27 +1115,21 @@ function resampleLinear(tSrc, vSrc, tDst) {
 function buildLiveData(p) {
   if (p.traces.length === 0) return [new Float64Array(0)];
 
-  // plotNow = min(newest per source) so no source shows a blank right edge.
   const plotNow = computePlotNow(p);
-
   const t0 = p.xRange ? p.xRange[0] : plotNow - windowSec;
   const t1 = p.xRange ? p.xRange[1] : plotNow;
 
-  // Pixel-adaptive LTTB target: 2× plot width so zooming in automatically
-  // raises the effective sample cap and reveals full resolution.
-  const targetPts = Math.max(LTTB_MIN, ((p.uplot ? p.uplot.width : p.div.clientWidth) || 600) * 2);
+  const isRolling = !p.xRange;
 
   // When zoomed, prefer server-fetched hi-res data if it covers this exact range.
   if (p.xRange) {
     const zd = zoomData[p.id];
     if (zd && Math.abs(zd.t0 - t0) < 1e-9 && Math.abs(zd.t1 - t1) < 1e-9) {
-      return buildDataFromFetched(p, zd.signals, targetPts);
+      return buildDataFromFetched(p, zd.signals, Math.max(LTTB_MIN, ((p.uplot ? p.uplot.width : p.div.clientWidth) || 600) * 2));
     }
   }
 
-  // Slice all traces once; pick the master time grid using configured samplingRate
-  // as the primary criterion (unambiguous, independent of buffer fill / trace order).
-  // Fall back to raw sample count for signals without a configured rate.
+  // Slice all traces; pick master by sampling rate then count.
   const slices = {};
   let masterKey = p.traces[0], masterCount = -1, masterRate = -1;
   for (const key of p.traces) {
@@ -1069,13 +1147,24 @@ function buildLiveData(p) {
   if (!masterRaw || masterRaw.t.length === 0)
     return [new Float64Array(0), ...p.traces.map(() => new Float64Array(0))];
 
-  // Decimate master with pixel-adaptive LTTB, use resulting grid for all others
+  // In rolling mode, Go backend already LTTB-decimated temporal signals to
+  // maxPushPoints (2000) and scalar points per tick are naturally limited.
+  // Skip JS-side LTTB entirely — just use the raw buffer data as-is.
+  // In zoomed mode, run pixel-adaptive LTTB for display quality.
+  let sharedT, masterV;
+  if (isRolling) {
+    sharedT = masterRaw.t;
+    masterV = masterRaw.v;
+  } else {
+    const targetPts = Math.max(LTTB_MIN, ((p.uplot ? p.uplot.width : p.div.clientWidth) || 600) * 2);
     const dec = lttb(masterRaw.t, masterRaw.v, targetPts);
-  const sharedT = dec.t;
+    sharedT = dec.t;
+    masterV = dec.v;
+  }
 
   const yArrays = [];
   for (const key of p.traces) {
-    if (key === masterKey) { yArrays.push(dec.v); continue; }
+    if (key === masterKey) { yArrays.push(masterV); continue; }
     const sl = slices[key];
     if (!sl || sl.t.length === 0) { yArrays.push(new Float64Array(sharedT.length)); continue; }
     yArrays.push(resampleLinear(sl.t, sl.v, sharedT));
@@ -1759,7 +1848,7 @@ function addPlot() {
   document.getElementById('plot-grid').appendChild(card);
 
   const plotBody = card.querySelector('#pbody-' + id);
-  const p = { id, traces: [], div: plotBody, needsRedraw: false, xRange: null, uplot: null, ro: null };
+  const p = { id, traces: [], div: plotBody, needsRedraw: false, xRange: null, uplot: null, ro: null, lastDataGen: -1 };
   plots.push(p);
   // uPlot creation is handled by applyLayout (batch, after DOM settles).
   return id;
@@ -1822,8 +1911,8 @@ function deletePlot(plotId) {
    Render loop
    ════════════════════════════════════════════════════════════════ */
 let _dbgTick = 0;
+let _dataGen = 0;  // incremented each time new data arrives
 function renderDirtyPlots() {
-  const inTrigMode = trig.enabled && trig.snapshot !== null;
 
   // Diagnostic: every ~5 s print buffer state to the browser console.
   // Open DevTools → Console to see timestamps and sizes.
@@ -1872,20 +1961,37 @@ function renderDirtyPlots() {
   }
 
   // Rolling-window plots: mark dirty every frame for smooth continuous scrolling.
-  // setScale is called AFTER setData inside the rebuild loop so the viewport and
-  // data slice are always computed with the same plotNow anchor.
+  // When no new data arrived since the last render, only advance the viewport
+  // via setScale instead of rebuilding all data arrays (much cheaper).
   if (!trig.enabled && !globalPause) {
     plots.forEach(p => {
-      if (!p.uplot || p.xRange) return;
+      if (!p.uplot || p.xRange || p.traces.length === 0) return;
       p.needsRedraw = true;
     });
   }
 
   plots.forEach(p => {
-    if (!p.needsRedraw || !p.uplot) return;
-    p.needsRedraw = false;
+    if (!p.needsRedraw || !p.uplot || p.traces.length === 0) return;
 
-    const data = buildUPlotData(p, inTrigMode);
+    const inTrigModeNow = trig.enabled && trig.snapshot !== null;
+    const isRolling = !trig.enabled && !p.xRange;
+
+    // Fast path: rolling-window plot with no new data — just shift viewport
+    // anchored to buffer timestamps so the x-range only advances when
+    // signal data actually moves forward.
+    if (isRolling && _dataGen === p.lastDataGen && p.uplot.data && p.uplot.data[0] && p.uplot.data[0].length > 0) {
+      p.needsRedraw = false;
+      zoomGuard = true;
+      const plotNow = computePlotNow(p);
+      p.uplot.setScale('x', { min: plotNow - windowSec, max: plotNow });
+      zoomGuard = false;
+      return;
+    }
+
+    p.needsRedraw = false;
+    p.lastDataGen = _dataGen;
+
+    const data = buildUPlotData(p, inTrigModeNow);
 
     // setData internally triggers the setScale hook in uPlot (it reaffirms the
     // current scale even with auto:false).  Keep zoomGuard raised across the
@@ -1896,9 +2002,9 @@ function renderDirtyPlots() {
     p.uplot.setData(data);
 
     // Re-apply the x-scale after setData so the viewport stays correct.
-    if (trig.enabled && !inTrigMode) {
+    if (trig.enabled && !inTrigModeNow) {
       // Armed / waiting for trigger: keep the current scale frozen.
-    } else if (inTrigMode) {
+    } else if (inTrigModeNow) {
       const preS = trig.snapshot._preS !== undefined ? trig.snapshot._preS : trigPreSec();
       const postS = trig.snapshot._postS !== undefined ? trig.snapshot._postS : trigPostSec();
       p.uplot.setScale('x', {

Client/WebUI/static/worker.js

@@ -0,0 +1,283 @@
+'use strict';
+/* ════════════════════════════════════════════════════════════════
+   Web Worker – buffer management, binary parsing, LTTB
+   ════════════════════════════════════════════════════════════════ */
+
+const TEMPORAL_CAP = 600_000;
+const DEFAULT_CAP = 10_000;
+
+// Circular buffers: key → {t:Float64Array, v:Float64Array, head, size, cap}
+const buffers = {};
+
+function makeBuffer(cap) {
+  return { t: new Float64Array(cap), v: new Float64Array(cap), head: 0, size: 0, cap };
+}
+function pushBuffer(buf, t, v) {
+  buf.t[buf.head] = t; buf.v[buf.head] = v;
+  buf.head = (buf.head + 1) % buf.cap;
+  if (buf.size < buf.cap) buf.size++;
+}
+
+// ─── Binary frame parser ─────────────────────────────────────────────
+// Format (little-endian):
+//   uint8  version (1)
+//   uint8  sourceIdLen
+//   UTF-8  sourceId
+//   uint32 numSignals
+//   for each signal:
+//     uint16 keyLen
+//     UTF-8  key (relative to source)
+//     uint32 pairCount N
+//     float64[N] t values
+//     float64[N] v values
+function parseBinaryFrame(buf) {
+  const dv = new DataView(buf);
+  let off = 0;
+
+  if (dv.getUint8(off) !== 1) { console.warn('[worker] bad binary version'); return; }
+  off += 1;
+
+  const srcIdLen = dv.getUint8(off); off += 1;
+  const srcId = new TextDecoder().decode(new Uint8Array(buf, off, srcIdLen));
+  off += srcIdLen;
+
+  const prefix = srcId + ':';
+  const numSigs = dv.getUint32(off, true); off += 4;
+
+  for (let s = 0; s < numSigs; s++) {
+    const keyLen = dv.getUint16(off, true); off += 2;
+    const key = new TextDecoder().decode(new Uint8Array(buf, off, keyLen));
+    off += keyLen;
+
+    const fullKey = prefix + key;
+    const n = dv.getUint32(off, true); off += 4;
+
+    let bufObj = buffers[fullKey];
+    if (!bufObj) {
+      // Auto-create buffer with reasonable capacity
+      const cap = n > 100 ? TEMPORAL_CAP : DEFAULT_CAP;
+      bufObj = makeBuffer(cap);
+      buffers[fullKey] = bufObj;
+    }
+
+    // Read t values
+    for (let i = 0; i < n; i++) {
+      const t = dv.getFloat64(off, true); off += 8;
+      const v = dv.getFloat64(off + n * 8, true); // v array starts after t array
+      pushBuffer(bufObj, t, v);
+    }
+    off += n * 8; // skip v array (already read inline above)
+  }
+}
+
+// ─── Range slice from circular buffer ────────────────────────────────
+function getBufferSliceRange(bufObj, t0, t1) {
+  const { cap, size, head } = bufObj;
+  if (size === 0) return { t: new Float64Array(0), v: new Float64Array(0) };
+  const start = (size === cap) ? head : 0;
+  const physAt = k => (start + k) % cap;
+
+  let lo = 0, hi = size;
+  while (lo < hi) { const m = (lo + hi) >>> 1; if (bufObj.t[physAt(m)] < t0) lo = m + 1; else hi = m; }
+  const kStart = lo;
+  lo = kStart; hi = size;
+  while (lo < hi) { const m = (lo + hi) >>> 1; if (bufObj.t[physAt(m)] <= t1) lo = m + 1; else hi = m; }
+  const kEnd = lo, len = kEnd - kStart;
+  if (len <= 0) return { t: new Float64Array(0), v: new Float64Array(0) };
+
+  const outT = new Float64Array(len), outV = new Float64Array(len);
+  const physStart = physAt(kStart), tail = cap - physStart;
+  if (tail >= len) {
+    outT.set(bufObj.t.subarray(physStart, physStart + len));
+    outV.set(bufObj.v.subarray(physStart, physStart + len));
+  } else {
+    outT.set(bufObj.t.subarray(physStart, physStart + tail));
+    outT.set(bufObj.t.subarray(0, len - tail), tail);
+    outV.set(bufObj.v.subarray(physStart, physStart + tail));
+    outV.set(bufObj.v.subarray(0, len - tail), tail);
+  }
+  return { t: outT, v: outV };
+}
+
+// ─── LTTB decimation ─────────────────────────────────────────────────
+function lttb(t, v, threshold) {
+  const len = t.length;
+  if (len <= threshold || threshold < 3) return { t, v };
+  const outT = new Float64Array(threshold), outV = new Float64Array(threshold);
+  outT[0] = t[0]; outV[0] = v[0];
+  outT[threshold - 1] = t[len - 1]; outV[threshold - 1] = v[len - 1];
+  const every = (len - 2) / (threshold - 2);
+  let a = 0;
+  for (let i = 0; i < threshold - 2; i++) {
+    const avgS = Math.floor((i + 1) * every) + 1, avgE = Math.min(Math.floor((i + 2) * every) + 1, len);
+    let avgT = 0, avgV = 0, n = 0;
+    for (let j = avgS; j < avgE; j++) { avgT += t[j]; avgV += v[j]; n++; }
+    if (n) { avgT /= n; avgV /= n; }
+    const rS = Math.floor(i * every) + 1, rE = Math.min(Math.floor((i + 1) * every) + 1, len);
+    let maxA = -1, next = rS;
+    const aT = t[a], aV = v[a];
+    for (let j = rS; j < rE; j++) {
+      const area = Math.abs((aT - avgT) * (v[j] - aV) - (aT - t[j]) * (avgV - aV));
+      if (area > maxA) { maxA = area; next = j; }
+    }
+    outT[i + 1] = t[next]; outV[i + 1] = v[next]; a = next;
+  }
+  return { t: outT, v: outV };
+}
+
+// ─── Linear resampling ───────────────────────────────────────────────
+function resampleLinear(tSrc, vSrc, tDst) {
+  const n = tDst.length;
+  const out = new Float64Array(n);
+  if (tSrc.length === 0) return out;
+  if (tSrc.length === 1) { out.fill(vSrc[0]); return out; }
+  let j = 0;
+  for (let i = 0; i < n; i++) {
+    const td = tDst[i];
+    while (j < tSrc.length - 2 && tSrc[j + 1] < td) j++;
+    if (td <= tSrc[0]) { out[i] = vSrc[0]; }
+    else if (td >= tSrc[tSrc.length - 1]) { out[i] = vSrc[vSrc.length - 1]; }
+    else {
+      const t0 = tSrc[j], t1 = tSrc[j + 1];
+      const frac = (td - t0) / (t1 - t0);
+      out[i] = vSrc[j] + frac * (vSrc[j + 1] - vSrc[j]);
+    }
+  }
+  return out;
+}
+
+// ─── Master time grid selection ──────────────────────────────────────
+// samplingRates: key → rate (Hz), provided by main thread on init
+const samplingRates = {};
+
+function pickMasterKey(keys) {
+  let bestKey = keys[0], bestRate = -1;
+  for (const k of keys) {
+    const rate = samplingRates[k] || 0;
+    if (rate > bestRate) { bestRate = rate; bestKey = k; }
+  }
+  return bestKey;
+}
+
+// ─── Build uPlot-compatible data arrays ──────────────────────────────
+function buildRenderData(keys, t0, t1, targetPts) {
+  if (!keys || keys.length === 0) return [new Float64Array(0)];
+
+  const slices = {};
+  let masterKey = pickMasterKey(keys), masterCount = -1;
+
+  for (const key of keys) {
+    const bufObj = buffers[key];
+    if (!bufObj || bufObj.size === 0) continue;
+    const sl = getBufferSliceRange(bufObj, t0, t1);
+    slices[key] = sl;
+    if (sl.t.length > masterCount) { masterCount = sl.t.length; masterKey = key; }
+  }
+
+  const masterRaw = slices[masterKey];
+  if (!masterRaw || masterRaw.t.length === 0)
+    return [new Float64Array(0), ...keys.map(() => new Float64Array(0))];
+
+  const dec = lttb(masterRaw.t, masterRaw.v, targetPts);
+  const sharedT = dec.t;
+  const yArrays = [];
+
+  for (const key of keys) {
+    if (key === masterKey) { yArrays.push(dec.v); continue; }
+    const sl = slices[key];
+    if (!sl || sl.t.length === 0) { yArrays.push(new Float64Array(sharedT.length)); continue; }
+    yArrays.push(resampleLinear(sl.t, sl.v, sharedT));
+  }
+
+  const result = [sharedT, ...yArrays];
+  // Transfer ownership of the Float64Arrays to main thread
+  const transferList = result.map(a => a.buffer);
+  return { data: result, transfer: transferList };
+}
+
+// ─── Message handler ─────────────────────────────────────────────────
+self.onmessage = function(e) {
+  const msg = e.data;
+
+  switch (msg.type) {
+    case 'initSignals': {
+      // {signals: [{key, cap}]}
+      const sigs = msg.signals || [];
+      sigs.forEach(s => {
+        if (!buffers[s.key]) {
+          buffers[s.key] = makeBuffer(s.cap || DEFAULT_CAP);
+        }
+        if (s.samplingRate !== undefined) {
+          samplingRates[s.key] = s.samplingRate;
+        }
+      });
+      break;
+    }
+
+    case 'binaryData': {
+      // {buffer: ArrayBuffer} — transferred from main thread
+      parseBinaryFrame(msg.buffer);
+      self.postMessage({ type: 'dataReady' });
+      break;
+    }
+
+    case 'requestData': {
+      // {id, t0, t1, targetPts, keys}
+      const { id, t0, t1, targetPts, keys } = msg;
+      const { data, transfer } = buildRenderData(keys, t0, t1, targetPts);
+      self.postMessage({ type: 'renderData', id, data }, transfer);
+      break;
+    }
+
+    case 'clearSource': {
+      const prefix = msg.prefix;
+      Object.keys(buffers).forEach(k => {
+        if (k.startsWith(prefix)) delete buffers[k];
+      });
+      Object.keys(samplingRates).forEach(k => {
+        if (k.startsWith(prefix)) delete samplingRates[k];
+      });
+      break;
+    }
+
+    case 'getBufferNow': {
+      // Returns newest timestamp across given keys
+      const keys = msg.keys || [];
+      let latest = -Infinity;
+      keys.forEach(key => {
+        const bufObj = buffers[key];
+        if (bufObj && bufObj.size > 0) {
+          const t = bufObj.t[(bufObj.head - 1 + bufObj.cap) % bufObj.cap];
+          if (t > latest) latest = t;
+        }
+      });
+      self.postMessage({ type: 'bufferNow', id: msg.id, now: isFinite(latest) ? latest : null });
+      break;
+    }
+
+    case 'getBufferForTrig': {
+      // Returns full buffer contents for a single key (used for trigger check)
+      const key = msg.key;
+      const bufObj = buffers[key];
+      if (!bufObj || bufObj.size === 0) {
+        self.postMessage({ type: 'trigBuf', id: msg.id, key, size: 0 });
+        break;
+      }
+      // Copy out all data
+      const { cap, size, head } = bufObj;
+      const start = (size === cap) ? head : 0;
+      const t = new Float64Array(size), v = new Float64Array(size);
+      const physAt = k => (start + k) % cap;
+      for (let i = 0; i < size; i++) {
+        const p = physAt(i);
+        t[i] = bufObj.t[p];
+        v[i] = bufObj.v[p];
+      }
+      self.postMessage({
+        type: 'trigBuf', id: msg.id, key, size,
+        t, v
+      }, [t.buffer, v.buffer]);
+      break;
+    }
+  }
+};

Client/WebUI/udpclient.go

@@ -10,6 +10,7 @@ const (
 	silenceTimeout = 5 * time.Second
 	reconnectDelay = 2 * time.Second
 	readBufSize    = 65536
+	udpRcvBufSize  = 8 * 1024 * 1024 // 8 MB OS receive buffer — absorbs bursts at high data rates
 )
 
 // UDPClient manages the UDP connection to one MARTe2 streamer source.
@@ -71,6 +72,11 @@ func (u *UDPClient) runSession() error {
 	}
 	defer conn.Close()
 
+	// Increase OS receive buffer to reduce kernel-level packet drops at high data rates.
+	if err := conn.SetReadBuffer(udpRcvBufSize); err != nil {
+		log.Printf("[%s] udp: SetReadBuffer: %v (proceeding with OS default)", u.sourceID, err)
+	}
+
 	serverAddr, err := net.ResolveUDPAddr("udp4", u.serverAddr)
 	if err != nil {
 		return err