iotlcm/worker-svr/internal/producer/monthlyusagehistory/blocked_zombie_test.go

package monthlyusagehistory

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

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

// Pause the real producer after its first enqueue, before MarkProducerDone.
type interruptedScanner struct {
	*datasetScanner
	paused chan struct{}
}

func TestInterruptedRunReclamationGuards(t *testing.T) {
	for _, mode := range []string{"serial", "channel"} {
		for _, status := range []string{"producing", "running"} {
			for _, guard := range []string{"no_lock", "foreign_lock", "no_current_run", "same_run", "producer_done", "replay"} {
				t.Run(mode+"/"+status+"/"+guard, func(t *testing.T) {
					ctx := contextForTest(t)
					store := testStore(t)
					p := testProducer(t, scannerFor(nil, 0), store, Config{ScanMode: mode})
					meta := zset.RunMeta{RunID: "guarded", Status: status, ProducerDone: guard == "producer_done"}
					if guard == "replay" {
						meta.SyncType = "replay"
					}
					if err := store.CreateRun(ctx, meta); err != nil {
						t.Fatal(err)
					}
					current := "fresh"
					owner := p.owner
					if guard == "same_run" {
						current = meta.RunID
					}
					if guard == "no_current_run" {
						current = ""
					}
					if guard == "foreign_lock" {
						owner = "other-owner"
					}
					if guard != "no_lock" {
						if ok, err := store.AcquireLockOrTakeover(ctx, p.config.RunLockKey, owner, current, p.config.RunLockTTL); err != nil || !ok {
							t.Fatalf("acquire=%t err=%v", ok, err)
						}
					}
					key, _ := store.RunMetaKey(meta.RunID)
					before, err := store.RedisClient().Do(ctx, "HGETALL", key)
					if err != nil {
						t.Fatal(err)
					}
					pending, err := p.hasPendingTasks(ctx, current)
					if err != nil || pending != (guard != "same_run") {
						t.Fatalf("guard=%s pending=%t err=%v", guard, pending, err)
					}
					after, err := store.RedisClient().Do(ctx, "HGETALL", key)
					if err != nil || !reflect.DeepEqual(before.MapStrStr(), after.MapStrStr()) {
						t.Fatalf("guarded metadata changed: before=%v after=%v err=%v", before, after, err)
					}
				})
			}
		}
	}
}

func (s *interruptedScanner) Scan(ctx context.Context, last, upper int64, size int) ([]Card, error) {
	if last > 0 {
		close(s.paused)
		<-ctx.Done()
		return nil, ctx.Err()
	}
	return s.datasetScanner.Scan(ctx, last, upper, size)
}

func (s *interruptedScanner) ScanChannel(ctx context.Context, key string, last int64, size int) ([]Card, error) {
	if last > 0 {
		close(s.paused)
		<-ctx.Done()
		return nil, ctx.Err()
	}
	return s.datasetScanner.ScanChannel(ctx, key, last, size)
}

func TestBlockedInterruptedProducerRecovery(t *testing.T) {
	for _, mode := range []string{"serial", "channel"} {
		for _, backlog := range []string{"none", "queued", "leased", "waiting"} {
			t.Run(mode+"/"+backlog, func(t *testing.T) {
				ctx := contextForTest(t)
				store := testStore(t)
				scanner := &interruptedScanner{datasetScanner: scannerFor([]Card{{ID: 1, ChanKey: "blocked"}, {ID: 2, ChanKey: "blocked"}}, 2), paused: make(chan struct{})}
				old := testProducer(t, scanner, store, Config{ScanMode: mode, PageSize: 1})
				oldCtx, interrupt := context.WithCancel(ctx)
				defer interrupt()
				result := make(chan error, 1)
				go func() { result <- old.Produce(oldCtx, scopedMessage("interrupted")) }()
				select {
				case <-scanner.paused:
				case err := <-result:
					t.Fatalf("producer exited before pause: %v", err)
				case <-ctx.Done():
					t.Fatal(ctx.Err())
				}
				claim := func(channel string) zset.ClaimedTask {
					t.Helper()
					tasks, err := store.ClaimRun(ctx, "interrupted", channel, time.Now().Add(time.Second), time.Now().Add(time.Minute), 1)
					if err != nil || len(tasks) != 1 {
						t.Fatalf("claim %s=%v err=%v", channel, tasks, err)
					}
					return tasks[0]
				}
				if n, err := store.BlockRunScopedAsFailed(ctx, claim("blocked"), zset.TaskOutcome{Code: 797, ErrCode: "S00004"}); err != nil || n != 1 {
					t.Fatalf("blocked=%d err=%v", n, err)
				}
				var remaining zset.ClaimedTask
				if backlog != "none" {
					tasks := []zset.Task{{TaskID: "remaining", RunID: "interrupted", ChanKey: "other"}}
					if n, err := store.EnqueueRunBatch(ctx, "interrupted", "other", tasks, time.Now()); err != nil || n != 1 {
						t.Fatalf("enqueue=%d err=%v", n, err)
					}
					if backlog != "queued" {
						remaining = claim("other")
					}
					if backlog == "waiting" {
						if ok, err := store.FinalizeRunTask(ctx, remaining, zset.TaskOutcome{Retry: true}); err != nil || !ok {
							t.Fatalf("defer=%t err=%v", ok, err)
						}
					}
				}
				before, err := store.GetRunMeta(ctx, "interrupted")
				if err != nil || before.Status != "running" || before.ProducerDone || before.FailedCount != 1 {
					t.Fatalf("blocked metadata=%+v err=%v", before, err)
				}
				fresh := testProducer(t, scannerFor([]Card{{ID: 3, ChanKey: "fresh"}}, 3), store, Config{ScanMode: mode})
				if err := fresh.Produce(ctx, scopedMessage("fresh")); !errors.Is(err, ErrAlreadyRunning) {
					t.Fatalf("live producer lock must prevent reclamation: %v", err)
				}
				requireMeta(t, store, "interrupted", false, before.ActiveCount)
				interrupt()
				if err := <-result; !errors.Is(err, context.Canceled) {
					t.Fatalf("interruption=%v", err)
				}
				lock, err := store.RedisClient().Do(ctx, "EXISTS", old.config.RunLockKey)
				if err != nil || lock.Int() != 0 {
					t.Fatalf("old lock=%v err=%v", lock, err)
				}
				err = fresh.Produce(ctx, scopedMessage("fresh"))
				if backlog == "none" && err != nil || backlog != "none" && !errors.Is(err, ErrAlreadyRunning) {
					t.Fatalf("fresh start with %s backlog: %v", backlog, err)
				}
				after, err := store.GetRunMeta(ctx, "interrupted")
				if err != nil || !after.ProducerDone || after.Status != "running" || after.ActiveCount != before.ActiveCount || after.InflightCount != before.InflightCount || after.NextRoundCount != before.NextRoundCount || after.FailedCount != before.FailedCount {
					t.Fatalf("reclaimed metadata=%+v before=%+v err=%v", after, before, err)
				}
				if backlog != "none" {
					if _, err := store.GetRunMeta(ctx, "fresh"); !errors.Is(err, zset.ErrRunNotFound) {
						t.Fatalf("backlog must prevent creating fresh run: %v", err)
					}
					if complete, err := store.CompleteRunIfIdle(ctx, "interrupted"); err != nil || complete {
						t.Fatalf("premature completion=%t err=%v", complete, err)
					}
					if backlog == "waiting" {
						if advanced, err := store.AdvanceRunRound(ctx, "interrupted"); err != nil || !advanced {
							t.Fatalf("advance=%t err=%v", advanced, err)
						}
					}
					if backlog != "leased" {
						remaining = claim("other")
					}
					if ok, err := store.FinalizeRunTask(ctx, remaining, zset.TaskOutcome{Success: true}); err != nil || !ok {
						t.Fatalf("drain=%t err=%v", ok, err)
					}
				}
				if complete, err := store.CompleteRunIfIdle(ctx, "interrupted", time.Hour); err != nil || !complete {
					t.Fatalf("normal completion=%t err=%v", complete, err)
				}
				closed, err := store.GetRunMeta(ctx, "interrupted")
				if err != nil || closed.Status != "completed_with_failures" || closed.FailedCount != 1 || closed.ActiveCount != 0 || closed.InflightCount != 0 || closed.NextRoundCount != 0 {
					t.Fatalf("closed metadata=%+v err=%v", closed, err)
				}
				metaKey, _ := store.RunMetaKey("interrupted")
				failedKey, _ := store.FailedKey("interrupted")
				stateKey, _ := store.TaskStateKey("interrupted")
				channelsKey, _ := store.RunChannelsKey("interrupted")
				for _, key := range []string{metaKey, failedKey, stateKey, channelsKey} {
					ttl, err := store.RedisClient().Do(ctx, "PTTL", key)
					if err != nil || ttl.Int64() <= 0 || ttl.Int64() > time.Hour.Milliseconds() {
						t.Fatalf("retention %s=%v err=%v", key, ttl, err)
					}
				}
				if backlog != "none" {
					if err := fresh.Produce(ctx, scopedMessage("fresh")); err != nil {
						t.Fatalf("fresh start after draining: %v", err)
					}
				}
				requireMeta(t, store, "fresh", true, 1)
				active, err := store.ActiveRuns(ctx)
				if err != nil || len(active) != 1 || active[0] != "fresh" {
					t.Fatalf("only closed run must leave active index: %v err=%v", active, err)
				}
			})
		}
	}
}