iotlcm/worker-svr/internal/fullsync/supply/observation_test.go

package supply

import (
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"testing"
	"time"
)

func TestSnapshotLimitPreservesSummaryAndRanking(t *testing.T) {
	keys := make([]string, 700)
	for i := range keys {
		keys[i] = fmt.Sprintf("%04d-%s", i, strings.Repeat("x", 200))
	}
	start := time.Unix(100, 0)
	o := New("attempt", start, keys)
	for i, k := range keys {
		o.Event(k, "global_capacity", 1, false, false, start)
		o.Apply(k, o.Views()[k], Sample{Known: true, Eligible: true}, start.Add(time.Second))
		o.Apply(k, o.Views()[k], Sample{Known: true, Eligible: true}, start.Add(time.Duration(1000+i)*time.Millisecond+time.Second))
	}
	body, err := o.Encode()
	if err != nil {
		t.Fatal(err)
	}
	var s Snapshot
	if err = json.Unmarshal(body, &s); err != nil {
		t.Fatal(err)
	}
	if len(body) > 256*1024 || !s.Truncated || len(s.Channels) != 100 || s.Summary.Targets != 700 || s.Channels[0].Key != keys[699] {
		t.Fatalf("invalid bounded snapshot bytes=%d %+v", len(body), s.Summary)
	}
	restored := New("new", start.Add(time.Hour), keys)
	restored.Restore(s)
	r := restored.Snapshot()
	if !r.Partial || !r.Truncated || r.Summary.TotalMS != s.Summary.TotalMS {
		t.Fatalf("lost historical totals %+v %+v", s.Summary, r.Summary)
	}
}

func TestConfirmedIntervalsAndStages(t *testing.T) {
	start := time.Unix(100, 0)
	o := New("attempt", start, []string{"a", "empty"})
	o.Event("a", "global_capacity", 2, false, false, start)
	tick := func(n int) {
		v := o.Views()["a"]
		o.Apply("a", v, Sample{Known: true, Eligible: true}, start.Add(time.Duration(n)*time.Second))
	}
	tick(1)
	tick(2)
	o.Event("a", "redis_write", 2, false, false, start.Add(2*time.Second))
	tick(3)
	s := o.Snapshot()
	c := s.Channels[0]
	if c.Startup.Count != 1 || c.Startup.TotalMS != 2000 || c.Stages["global_capacity"] != 1000 || c.Stages["redis_write"] != 1000 {
		t.Fatalf("unexpected %+v", c)
	}
	o.Processing("a", 1)
	tick(4)
	tick(5)
	if o.Snapshot().Channels[0].Startup.TotalMS != 2000 {
		t.Fatal("processing counted as idle")
	}
}
func TestUncertainAndGap(t *testing.T) {
	start := time.Unix(100, 0)
	o := New("a", start, []string{"a"})
	v := o.Views()["a"]
	o.Event("a", "sql_query", 0, false, false, start)
	o.Apply("a", v, Sample{Known: true, Eligible: true}, start.Add(time.Second))
	o.Apply("a", o.Views()["a"], Sample{Known: false}, start.Add(2*time.Second))
	o.Apply("a", o.Views()["a"], Sample{Known: true, Eligible: true}, start.Add(6*time.Second))
	s := o.Snapshot()
	if !s.Partial || s.Channels[0].UnknownMS != 1000 || s.Channels[0].MissingSamples != 5 || s.Summary.Confirmed != 0 {
		t.Fatalf("unexpected %+v", s)
	}
}
func TestSupplyExclusions(t *testing.T) {
	for _, tc := range []struct {
		name       string
		sample     Sample
		processing bool
	}{{"stock", Sample{Known: true, Eligible: true, Ready: 1}, false}, {"lease", Sample{Known: true, Eligible: true, Leases: 1}, false}, {"ended_round_or_blocked", Sample{Known: true, Eligible: false}, false}, {"consumer", Sample{Known: true, Eligible: true}, true}, {"unknown", Sample{Eligible: true}, false}} {
		t.Run(tc.name, func(t *testing.T) {
			start := time.Unix(100, 0)
			o := New("a", start, []string{"a"})
			o.Event("a", "global_capacity", 100, false, false, start)
			if tc.processing {
				o.Processing("a", 1)
			}
			for i := 1; i <= 3; i++ {
				o.Apply("a", o.Views()["a"], tc.sample, start.Add(time.Duration(i)*time.Second))
			}
			if o.Snapshot().Summary.Confirmed != 0 {
				t.Fatal("excluded sample counted as supply idle")
			}
		})
	}
}
func TestConcurrentEvents(t *testing.T) {
	start := time.Unix(100, 0)
	o := New("a", start, []string{"a"})
	var wg sync.WaitGroup
	for i := 0; i < 8; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for n := 0; n < 100; n++ {
				o.Event("a", "sql_query", 0, false, false, start)
				o.Processing("a", 1)
				o.Empty("a")
				o.Apply("a", o.Views()["a"], Sample{Known: true, Eligible: true}, start)
				o.Snapshot()
				o.Processing("a", -1)
			}
		}()
	}
	wg.Wait()
}

func TestEmptyNeverConfirms(t *testing.T) {
	start := time.Unix(100, 0)
	o := New("a", start, []string{"a"})
	for i := 1; i <= 3; i++ {
		o.Apply("a", o.Views()["a"], Sample{Known: true, Eligible: true}, start.Add(time.Duration(i)*time.Second))
	}
	o.Event("a", "empty", 0, false, true, start.Add(3*time.Second))
	s := o.Snapshot()
	if s.Summary.Confirmed != 0 || s.Channels[0].Unconfirmed.TotalMS != 2000 {
		t.Fatalf("unexpected %+v", s)
	}
}