iotlcm/internal/producer/dailyusage/supply_observation_test.go

package dailyusage

import (
	"reflect"
	"testing"
	"time"

	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
)

func TestLocalSupplyObservationFirstEventsAndStages(t *testing.T) {
	started := time.Now()
	o := &supplyObservation{started: started, channels: map[string]*supplyChannel{"a": {stage: "initializing", changed: started, stageDuration: map[string]time.Duration{}}}}
	o.event("a", "sql_query", 0, false, false, started.Add(10*time.Millisecond))
	o.event("a", "page_ready", 2, false, false, started.Add(20*time.Millisecond))
	o.event("a", "redis_write", 1, true, false, started.Add(30*time.Millisecond))
	o.event("a", "channel_capacity", 1, false, false, started.Add(40*time.Millisecond))
	o.event("a", "sql_query", 0, false, false, started.Add(50*time.Millisecond))
	o.event("a", "redis_write", 0, true, false, started.Add(60*time.Millisecond))
	o.event("a", "upper_bound", 0, false, true, started.Add(70*time.Millisecond))
	o.event("unknown", "ignored", 0, false, false, started)
	s := o.channels["a"]
	if s.firstQuery == nil || *s.firstQuery != 10 || s.firstEnqueue == nil || *s.firstEnqueue != 30 || !s.ended || s.pending != 0 || s.version != 7 || s.stageDuration["channel_capacity"] != 10*time.Millisecond || s.stageDuration["sql_query"] != 20*time.Millisecond {
		t.Fatalf("observation=%+v", s)
	}
}

func TestSupplyObservationAccumulatesSubMillisecondStages(t *testing.T) {
	started := time.Unix(0, 0)
	o := &supplyObservation{started: started, channels: map[string]*supplyChannel{"a": {stage: "redis_read", changed: started, stageDuration: map[string]time.Duration{}}}}
	for i := 1; i <= 100000; i++ {
		stage := "redis_read"
		if i%2 == 1 {
			stage = "redis_write"
		}
		o.event("a", stage, 0, false, i == 100000, started.Add(time.Duration(i)*500*time.Microsecond))
	}
	if got := o.channels["a"].stageMilliseconds(started.Add(time.Hour)); got["redis_read"] != 25000 || got["redis_write"] != 25000 {
		t.Fatalf("stageMs=%v, want 25000ms per stage", got)
	}
}

func TestSupplyObservationFinalFlushPreservesFraction(t *testing.T) {
	started := time.Unix(0, 0)
	o := &supplyObservation{started: started, channels: map[string]*supplyChannel{"a": {stage: "redis_write", changed: started, stageDuration: map[string]time.Duration{}}}}
	o.event("a", "redis_read", 0, false, false, started.Add(600*time.Microsecond))
	o.event("a", "redis_write", 0, false, false, started.Add(1200*time.Microsecond))
	s := o.channels["a"]
	if got := s.stageMilliseconds(started.Add(1800 * time.Microsecond)); got["redis_write"] != 1 || got["redis_read"] != 0 {
		t.Fatalf("final stageMs=%v", got)
	}
	if s.stageDuration["redis_write"] != 1200*time.Microsecond || s.stageDuration["redis_read"] != 600*time.Microsecond {
		t.Fatalf("fractional duration lost: %v", s.stageDuration)
	}
}

func TestSupplySamplesDistinguishReadyLeaseFutureAndClosedRound(t *testing.T) {
	store := testStore(t)
	ctx := contextForTest(t)
	p := testProducer(t, scannerFor(nil, 0), store, Config{})
	if err := store.CreateRun(ctx, zset.RunMeta{RunID: "supply"}); err != nil {
		t.Fatal(err)
	}
	active, _ := store.RoundActiveKey("supply", 1, "a")
	lease, _ := store.RoundLeaseKey("supply", 1, "a")
	_, _ = store.RedisClient().Do(ctx, "ZADD", active, time.Now().Add(-time.Minute).UnixMilli(), "ready", time.Now().Add(time.Hour).UnixMilli(), "future")
	_, _ = store.RedisClient().Do(ctx, "HSET", lease, "leased", "token")
	values, err := p.sampleSupply(ctx, "supply", "a")
	if err != nil || !reflect.DeepEqual(values, []int64{1, 1, 1, 1, 1}) {
		t.Fatalf("sample=%v err=%v", values, err)
	}
	meta, _ := store.RunMetaKey("supply")
	for _, args := range [][]any{{"blocked:a", 1}, {"blocked:a", 0, "producerDone", 1}, {"producerDone", 0, "currentRound", 2}, {"currentRound", 1, "syncType", "replay"}} {
		_, _ = store.RedisClient().Do(ctx, "HSET", append([]any{meta}, args...)...)
		values, err = p.sampleSupply(ctx, "supply", "a")
		if err != nil || values[0] != 1 || values[1] != 0 {
			t.Fatalf("ineligible sample=%v err=%v", values, err)
		}
	}
	values, err = p.sampleSupply(ctx, "missing", "a")
	if err != nil || values[0] != 0 || values[1] != 0 {
		t.Fatalf("missing sample=%v err=%v", values, err)
	}
}

func TestSupplySamplerTracksWaitingChannelsAndStops(t *testing.T) {
	store := testStore(t)
	ctx := contextForTest(t)
	p := testProducer(t, scannerFor(nil, 0), store, Config{})
	if err := store.CreateRun(ctx, zset.RunMeta{RunID: "sampling"}); err != nil {
		t.Fatal(err)
	}
	finish := p.startObservation(ctx, "sampling", []string{"a", "b"}, time.Now())
	defer finish()
	o := p.observation
	p.observe("a", "channel_capacity", 1, false, false)
	p.observe("b", "sql_slot", 0, false, false)
	waitUntil(t, ctx, func() bool {
		o.mu.Lock()
		defer o.mu.Unlock()
		return o.channels["a"].validSamples > 0 && o.channels["b"].validSamples > 0
	})
	p.observe("a", "cancelled", 0, false, true)
	p.observe("b", "cancelled", 0, false, true)
	o.mu.Lock()
	defer o.mu.Unlock()
	if o.channels["a"].stageDuration["channel_capacity"] < 900*time.Millisecond || o.channels["b"].stageDuration["sql_slot"] < 900*time.Millisecond {
		t.Fatalf("missing wait attribution: a=%+v b=%+v", o.channels["a"], o.channels["b"])
	}
}