iotlcm/internal/producer/monthlyusagehistory/parallel_regression_test.go

package monthlyusagehistory

import (
	"context"
	"errors"
	"testing"
	"time"

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

func TestChannelBackpressureDoesNotStarveHighIDChannel(t *testing.T) {
	for _, mode := range []string{"serial", "channel"} {
		t.Run(mode, func(t *testing.T) {
			store := testStore(t)
			ctx, cancel := context.WithCancel(contextForTest(t))
			defer cancel()
			scanner := scannerFor([]Card{{ID: 1, ChanKey: "slow"}, {ID: 2, ChanKey: "slow"}, {ID: 3, ChanKey: "slow"}, {ID: 1000, ChanKey: "fast"}}, 1000)
			p := testProducer(t, scanner, store, Config{ScanMode: mode, PageSize: 1, ScanConcurrency: 1, ChannelHighWaterMark: 1, GlobalHighWaterMark: 10, PollInterval: time.Millisecond})
			result := make(chan error, 1)
			started := time.Now()
			go func() { result <- p.Produce(ctx, scopedMessage("starvation")) }()
			slowKey, _ := store.RoundActiveKey("starvation", 1, "slow")
			fastKey, _ := store.RoundActiveKey("starvation", 1, "fast")
			waitUntil(t, ctx, func() bool { n, e := store.SizeKey(ctx, slowKey); return e == nil && n == 1 })
			if mode == "channel" {
				waitUntil(t, ctx, func() bool { n, e := store.SizeKey(ctx, fastKey); return e == nil && n == 1 })
			} else {
				// No consumer drains slow. The serial cursor cannot reach card 1000.
				time.Sleep(30 * time.Millisecond)
				if n, err := store.SizeKey(ctx, fastKey); err != nil || n != 0 {
					t.Fatalf("serial fast size=%d err=%v", n, err)
				}
			}
			slow, _ := store.SizeKey(ctx, slowKey)
			fast, _ := store.SizeKey(ctx, fastKey)
			t.Logf("mode=%s slowQueued=%d fastQueued=%d scanConcurrency=1 channelHWM=1 elapsedMs=%d", mode, slow, fast, time.Since(started).Milliseconds())
			cancel()
			if err := <-result; !errors.Is(err, context.Canceled) {
				t.Fatalf("cancellation err=%v", err)
			}
			requireMeta(t, store, "starvation", false, slow+fast)
		})
	}
}

func TestChannelPayloadRetainsPeriodPackageAndIdentifiers(t *testing.T) {
	store := testStore(t)
	ctx := contextForTest(t)
	card := Card{ID: 10, ICCID: "iccid", MSISDN: "msisdn", IMSI: "imsi", ChanKey: "a", PackageCode: "pkg-history"}
	p := testProducer(t, scannerFor([]Card{card}, card.ID), store, Config{PageSize: 1})
	message := scopedMessage("payload", "a")
	if err := p.Produce(ctx, message); err != nil {
		t.Fatal(err)
	}
	claimed, err := store.ClaimRun(ctx, message.RunID, "a", time.Now().Add(time.Second), time.Now().Add(time.Minute), 1)
	if err != nil || len(claimed) != 1 {
		t.Fatalf("claimed=%v err=%v", claimed, err)
	}
	task := claimed[0].Task
	if task.TaskID != zset.StableTaskID(message.RunID, FullSyncOperation, card.ID) || task.Operation != FullSyncOperation || task.QueryMonth != message.QueryMonth || task.PackageCode != card.PackageCode || task.ICCID != card.ICCID || task.MSISDN != card.MSISDN || task.IMSI != card.IMSI || task.TempKey != "sim_yd_v5" || task.Operator != 1 {
		t.Fatalf("task=%+v", task)
	}
}

func TestZombieWithBacklogDrainsBeforeCompletion(t *testing.T) {
	p, store := newZombieTestProducer(t)
	ctx := contextForTest(t)
	if err := store.CreateRun(ctx, zset.RunMeta{RunID: "zombie-backlog"}); err != nil {
		t.Fatal(err)
	}
	tasks := []zset.Task{{TaskID: "zombie-card", RunID: "zombie-backlog", ChanKey: "a"}}
	if _, err := store.EnqueueRunBatch(ctx, "zombie-backlog", "a", tasks, time.Now()); err != nil {
		t.Fatal(err)
	}
	if pending, err := p.hasPendingTasks(ctx, "new"); err != nil || !pending {
		t.Fatalf("backlog must retain backpressure: pending=%t err=%v", pending, err)
	}
	requireMeta(t, store, "zombie-backlog", true, 1)
	if complete, err := store.CompleteRunIfIdle(ctx, "zombie-backlog"); err != nil || complete {
		t.Fatalf("premature complete=%t err=%v", complete, err)
	}
	claimed, err := store.ClaimRun(ctx, "zombie-backlog", "a", time.Now().Add(time.Second), time.Now().Add(time.Minute), 1)
	if err != nil || len(claimed) != 1 {
		t.Fatalf("claimed=%v err=%v", claimed, err)
	}
	if ok, err := store.FinalizeRunTask(ctx, claimed[0], zset.TaskOutcome{Success: true}); err != nil || !ok {
		t.Fatalf("finalize=%t err=%v", ok, err)
	}
	if complete, err := store.CompleteRunIfIdle(ctx, "zombie-backlog", time.Hour); err != nil || !complete {
		t.Fatalf("complete=%t err=%v", complete, err)
	}
	meta, _ := store.RunMetaKey("zombie-backlog")
	ttl, err := store.RedisClient().Do(ctx, "TTL", meta)
	if err != nil || ttl.Int() <= 0 {
		t.Fatalf("retention ttl=%v err=%v", ttl, err)
	}
	active, err := store.ActiveRuns(ctx)
	if err != nil || len(active) != 0 {
		t.Fatalf("active=%v err=%v", active, err)
	}
}