MARTe-Integrated-Components/Common/Client/go/wshub/packet_dt_loss_test.go

package wshub

import (
	"math"
	"testing"
	"time"

	"marte2/common/udpsprotocol"
)

// pktSignal is a 4-element array with no time signal and no declared sampling
// rate, i.e. the TimeModePacket path where dt has to be inferred from the gap
// between packets.
func pktSignal(name string) udpsprotocol.SignalInfo {
	return udpsprotocol.SignalInfo{
		Name:          name,
		TypeCode:      8, // float64
		NumDimensions: 1,
		NumRows:       4,
		NumCols:       1,
		TimeMode:      udpsprotocol.TimeModePacket,
		TimeSignalIdx: udpsprotocol.NoTimeSignal,
	}
}

// newPacketDtHub builds a Hub with a ring for one packet-timed array signal and
// returns both. It does not start Run(): buildBinaryDataMessageForSource is
// called directly so the timestamps it produces can be read back verbatim.
func newPacketDtHub(t *testing.T, sigName string) (*Hub, *sourceHubState, *sigRing) {
	t.Helper()
	h := NewHub()
	src := &sourceHubState{
		id:                "s1",
		signals:           []udpsprotocol.SignalInfo{pktSignal(sigName)},
		timeSigCalib:      map[string]float64{},
		lastPktNs:         map[string]int64{},
		lastFrameMeasured: map[string]float64{},
		lastFrameEndT:     map[string]float64{},
		gapEMA:            map[string]float64{},
	}
	rb := newSigRing(4096)
	h.rings["s1:"+sigName] = rb
	return h, src, rb
}

// packet builds a one-signal batch entry arriving at t0 with the given loss
// count; the values are irrelevant, only the timestamps are under test.
func packet(sigName string, at time.Time, lost uint32, n int) udpsprotocol.DataSample {
	vals := make([]float64, n)
	return udpsprotocol.DataSample{WallTime: at, Values: map[string][]float64{sigName: vals}, Lost: lost}
}

// ringTimes returns the timestamps written to the ring, in order.
func ringTimes(rb *sigRing) []float64 {
	rb.mu.RLock()
	defer rb.mu.RUnlock()
	out := make([]float64, 0, rb.size)
	start := (rb.head - rb.size + rb.cap) % rb.cap
	for i := 0; i < rb.size; i++ {
		out = append(out, rb.t[(start+i)%rb.cap])
	}
	return out
}

// A lost packet widens the inter-packet gap without adding elements to the
// packet that follows it. Dividing the gap by that packet's element count
// alone reports a period too long by exactly the number of packets missing,
// which walks the elements past their own end and into the range the next
// packet claims: they collide there, and the span they vacated stays empty.
func TestPacketDtIgnoresLostPacketWidening(t *testing.T) {
	const sig = "Wave"
	const n = 4
	const dt = 1 * time.Millisecond
	base := time.Unix(1700000000, 0)

	h, src, rb := newPacketDtHub(t, sig)

	// One clean packet establishes lastPktNs.
	h.buildBinaryDataMessageForSource(src, []udpsprotocol.DataSample{
		packet(sig, base, 0, n)})

	// The next producer packet is lost, so the one after it arrives a full
	// extra batch later and reports Lost=1.
	arrival := base.Add(2 * n * dt)
	h.buildBinaryDataMessageForSource(src, []udpsprotocol.DataSample{
		packet(sig, arrival, 1, n)})

	ts := ringTimes(rb)
	if len(ts) != n {
		t.Fatalf("ring holds %d points, want %d (the first packet is skipped: no gap yet)", len(ts), n)
	}
	got := ts[1] - ts[0]
	if !nearSec(got, dt.Seconds()) {
		t.Errorf("dt = %v s, want %v s (the gap spans two batches, not one)", got, dt.Seconds())
	}
	// Elements run forward from the packet's own arrival, so a doubled dt
	// would stretch this batch across two batch periods and into the range the
	// next packet claims.
	span := ts[len(ts)-1] - ts[0]
	if !nearSec(span, float64(n-1)*dt.Seconds()) {
		t.Errorf("batch spans %v s, want %v s: it overruns into the next packet's range",
			span, float64(n-1)*dt.Seconds())
	}
}

// nearSec compares two intervals in seconds. The hub carries timestamps as
// float64 seconds derived from UnixNano, whose spacing near the current epoch
// is a couple of hundred nanoseconds, so exact equality is not available. The
// defect under test moves the period by a factor of two, three orders of
// magnitude outside this tolerance.
func nearSec(got, want float64) bool { return math.Abs(got-want) <= 1e-6 }

// The correction must be driven by the reported loss and nothing else: with no
// packet missing the period still comes straight from the gap, so a producer
// that genuinely slows down is followed rather than second-guessed.
func TestPacketDtFollowsGapWhenNothingIsLost(t *testing.T) {
	const sig = "Wave"
	const n = 4
	base := time.Unix(1700000000, 0)

	h, src, rb := newPacketDtHub(t, sig)

	h.buildBinaryDataMessageForSource(src, []udpsprotocol.DataSample{
		packet(sig, base, 0, n)})

	// Same widened gap as the test above, but reported as no loss: the
	// producer really is running at half the rate.
	slowDt := 2 * time.Millisecond
	h.buildBinaryDataMessageForSource(src, []udpsprotocol.DataSample{
		packet(sig, base.Add(n*slowDt), 0, n)})

	ts := ringTimes(rb)
	if len(ts) != n {
		t.Fatalf("ring holds %d points, want %d", len(ts), n)
	}
	if got := ts[1] - ts[0]; !nearSec(got, slowDt.Seconds()) {
		t.Errorf("dt = %v s, want %v s: a real rate change must be followed", got, slowDt.Seconds())
	}
}