package wshub import ( "encoding/binary" "math" "testing" ) // decodeCaptureSpan pulls the time extent of one signal out of a v2 frame. func decodeCaptureSpan(t *testing.T, buf []byte, key string) (first, last float64, n int) { t.Helper() off := 1 + 8 + 8 + 8 nSig := int(binary.LittleEndian.Uint32(buf[off:])) off += 4 for i := 0; i < nSig; i++ { kl := int(binary.LittleEndian.Uint16(buf[off:])) off += 2 k := string(buf[off : off+kl]) off += kl cnt := int(binary.LittleEndian.Uint32(buf[off:])) off += 4 if k == key && cnt > 0 { first = math.Float64frombits(binary.LittleEndian.Uint64(buf[off:])) last = math.Float64frombits(binary.LittleEndian.Uint64(buf[off+(cnt-1)*8:])) n = cnt } off += cnt * 16 } return } // The rings only reach back over the window once they have rolled over at the // current bucket, which takes as long as the window itself — so a window widened // mid-run leaves the first captures asking for history the rings never stored. // The archive kept it, and the capture must come back whole. func TestCaptureBackfillsItsHeadFromTheArchive(t *testing.T) { h := NewHub() hw, key := newTestHistory(t, HistoryConfig{ WindowSec: 60, Decimation: 1, MinDiskFreeMB: -1, }, 1000) h.hist = hw // 20 s of 1 kSps, archived in full… ts, vs := ramp(1000, 0.001, 20000) hw.write(key, ts, vs) // …but a ring that only ever holds the last 5 s of it. rb := newSigRing(5000) rb.write(ts, vs) h.rings[key] = rb // A 15 s window, of which the ring has the newest third. const t0, t1 = 1005.0, 1020.0 buf := h.buildTriggerCapture(1015, 10, 5) if buf == nil { t.Fatal("no capture frame built") } first, last, n := decodeCaptureSpan(t, buf, key) if first > t0+0.05 { t.Errorf("capture starts at %.3f, want the window's start %.3f — the archive holds it", first, t0) } if last < t1-0.05 { t.Errorf("capture ends at %.3f, want %.3f", last, t1) } if n < 100 { t.Errorf("capture has %d points, too few for a 15 s window at 1 kSps", n) } // The join between the two sources must not break time order, or every // binary search over the capture — client-side and in the hold — misreads it. ct, _, ok := h.capture.slice(key, t0, t1) if !ok { t.Fatal("the hold declined the window it just published") } for i := 1; i < len(ct); i++ { if ct[i] < ct[i-1] { t.Fatalf("capture time goes backwards at %d: %.6f then %.6f", i, ct[i-1], ct[i]) } } } // A capture that neither source could fill must not answer for the stretch it is // missing: the client has to fall through to the archive instead of redrawing // the same hole on every zoom. func TestHoldDeclinesTheStretchACaptureNeverGot(t *testing.T) { h := NewHub() ts, vs := ramp(1000, 0.001, 20000) rb := newSigRing(5000) // the newest 5 s only, and no archive to fill from rb.write(ts, vs) h.rings["src:sig"] = rb if buf := h.buildTriggerCapture(1015, 10, 5); buf == nil { t.Fatal("no capture frame built") } if _, _, ok := h.capture.slice("src:sig", 1005, 1020); ok { t.Error("the hold answered for 15 s it only has the last 5 s of") } // What it does hold, it still serves. if _, _, ok := h.capture.slice("src:sig", 1016, 1019); !ok { t.Error("the hold declined a range well inside its data") } } // TestCaptureCoverageAcrossShots walks a whole acquisition the way Run() does — // ingest, retune, dueCapture, rearm — and reports how much of each window the // capture actually came back with. func TestCaptureCoverageAcrossShots(t *testing.T) { const ( key = "s1:Ch1" rate = 100e3 // scaled 10x down from the 1 MSps producer budget = 400_000 window = 120.0 prePct = 20.0 batchSec = 1.0 / 30.0 simSec = 900.0 ) h := NewHub() h.SetRingBudget(budget) h.rings[key] = newSigRing(ringCapInitial) h.trigger.SetConfig(trigConfig{signalKey: key, edge: "rising", threshold: 0, windowSec: window, prePercent: prePct, mode: "normal", holdoffSec: 0.2}) rateHz := float64(rate) nBatch := int(rateHz * batchSec) ts := make([]float64, nBatch) vs := make([]float64, nBatch) armed := false shots := 0 for now := 0.0; now < simSec; now += batchSec { for i := range ts { ts[i] = now + float64(i)/rateHz // 0.05 Hz sine: one rising zero crossing every 20 s. vs[i] = math.Sin(2 * math.Pi * 0.05 * ts[i]) } h.ingest(key, 1, ts, vs) h.retuneRings(now) // Arm once the stream is going, as a user would. if !armed && now > 5 { h.trigger.Arm() armed = true } if trigTime, pre, post, ok := h.trigger.dueCapture(now + batchSec); ok { buf := h.buildTriggerCapture(trigTime, pre, post) if buf == nil { t.Fatalf("shot at t=%.1f produced no frame", trigTime) } first, last, n := decodeCaptureSpan(t, buf, key) t0, t1 := trigTime-pre, trigTime+post _, ringSpan := h.rings[key].stats() shots++ t.Logf("shot %d fired t=%.1f window [%.1f,%.1f] got [%.1f,%.1f] "+ "= %.0f%% (%d pts, bucket %d, ring span %.1f s)", shots, trigTime, t0, t1, first, last, 100*(last-first)/(t1-t0), n, h.rings[key].bucketSize(), ringSpan) h.trigger.markTriggered(now + batchSec) } else if h.trigger.dueRearm(now + batchSec) { h.trigger.Arm() } } if shots < 3 { t.Fatalf("only %d shots in %.0f s", shots, simSec) } }