package server

import (
	"context"
	"errors"
	"fmt"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"go.uber.org/zap/zaptest/observer"

	"github.com/mem9-ai/drive9/pkg/backend"
	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/metrics"
)

func TestTenantWorkerShouldRateLimitedLog(t *testing.T) {
	t.Parallel()
	m := &tenantWorkerManager{
		lastRateLimitedLog: make(map[string]time.Time),
	}
	if !m.shouldRateLimitedLog("t1", "slot_busy", tenantWorkerSlotBusyLogInterval) {
		t.Fatal("first slot-busy log should be allowed")
	}
	if m.shouldRateLimitedLog("t1", "slot_busy", tenantWorkerSlotBusyLogInterval) {
		t.Fatal("second slot-busy log within interval should be rate-limited")
	}
	if !m.shouldRateLimitedLog("t1", "claim_empty_with_backlog", tenantWorkerSlotBusyLogInterval) {
		t.Fatal("different log name should not share rate limit")
	}
	if !m.shouldRateLimitedLog("t2", "slot_busy", tenantWorkerSlotBusyLogInterval) {
		t.Fatal("different tenant should not share rate limit")
	}
	m.lastRateLimitedLog["t1\x00slot_busy"] = time.Now().Add(-tenantWorkerSlotBusyLogInterval - time.Second)
	if !m.shouldRateLimitedLog("t1", "slot_busy", tenantWorkerSlotBusyLogInterval) {
		t.Fatal("slot-busy log after interval should be allowed again")
	}
}

func TestTenantWorkerSlotBusyAfterClaim(t *testing.T) {
	t.Parallel()
	m := &tenantWorkerManager{
		inflight:           make(map[string]int),
		kickPending:        make(map[string]pendingKick),
		kicks:              make(chan kickMsg, tenantKickQueueCapacity),
		lastRateLimitedLog: make(map[string]time.Time),
		opts:               TenantWorkerOptions{PerTenantConcurrency: 1},
	}
	m.opts.normalize()
	if !m.tryClaimTenantSlot("t1") {
		t.Fatal("first slot claim should succeed")
	}
	if m.tryClaimTenantSlot("t1") {
		t.Fatal("second slot claim should fail at concurrency=1")
	}
	if got := m.tenantSlotInflight("t1"); got != 1 {
		t.Fatalf("inflight = %d, want 1", got)
	}
	m.releaseTenantSlot("t1")
	if got := m.tenantSlotInflight("t1"); got != 0 {
		t.Fatalf("inflight after release = %d, want 0", got)
	}
	// Double-release must be a safe no-op: processKicked recover + defer
	// release both may run depending on panic timing.
	m.releaseTenantSlot("t1")
	if got := m.tenantSlotInflight("t1"); got != 0 {
		t.Fatalf("inflight after double release = %d, want 0", got)
	}
}

func TestTenantWorkerReleaseTenantSlotIdempotent(t *testing.T) {
	t.Parallel()
	// releaseTenantSlot is the shared primitive used by processKicked recover
	// (slotHeld path) and by target.release; it must tolerate double-call.
	m := &tenantWorkerManager{
		inflight:           make(map[string]int),
		kickPending:        make(map[string]pendingKick),
		kicks:              make(chan kickMsg, tenantKickQueueCapacity),
		lastRateLimitedLog: make(map[string]time.Time),
		opts:               TenantWorkerOptions{PerTenantConcurrency: 1},
	}
	m.opts.normalize()
	if !m.tryClaimTenantSlot("local") {
		t.Fatal("claim local slot")
	}
	m.releaseTenantSlot("local")
	if got := m.tenantSlotInflight("local"); got != 0 {
		t.Fatalf("after release inflight=%d, want 0", got)
	}
	if !m.tryClaimTenantSlot("local") {
		t.Fatal("slot should be free for a subsequent claim")
	}
	m.releaseTenantSlot("local")
}

func TestTenantWorkerProcessKickedPanicReleasesSlot(t *testing.T) {
	// No t.Parallel: mutates package-global processKickedAfterSlotClaimTestHook.
	//
	// End-to-end: processKicked claims the local tenant slot, the test hook
	// panics while slotHeld=true (before target.release owns the slot),
	// recover releases concurrency=1, and a subsequent claim succeeds.
	//
	// kickRef(local) with meta/pool nil accepts WorkFileGC without a fallback
	// backend; the hook fires before targetForRef, so no real backend is needed.
	m := &tenantWorkerManager{
		inflight:           make(map[string]int),
		kickPending:        make(map[string]pendingKick),
		kicks:              make(chan kickMsg, tenantKickQueueCapacity),
		lastRateLimitedLog: make(map[string]time.Time),
		opts:               TenantWorkerOptions{PerTenantConcurrency: 1},
	}
	m.opts.normalize()

	const panicTenant = tenantLocalID
	processKickedAfterSlotClaimTestHook = func(tenantID string) {
		if tenantID != panicTenant {
			t.Fatalf("hook tenant=%q, want %q", tenantID, panicTenant)
		}
		panic("test processKicked panic after slot claim")
	}
	t.Cleanup(func() { processKickedAfterSlotClaimTestHook = nil })

	// processKicked must recover; this call must not panic out of the test.
	m.processKicked(context.Background(), panicTenant, "org-panic-slot", WorkFileGC)

	if got := m.tenantSlotInflight(panicTenant); got != 0 {
		t.Fatalf("after processKicked panic inflight=%d, want 0 (slot must be free)", got)
	}
	if !m.tryClaimTenantSlot(panicTenant) {
		t.Fatal("slot should be claimable after panic recover released it")
	}
	m.releaseTenantSlot(panicTenant)

	// Live recover path must attribute panic on the tenant failure series.
	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	want := `drive9_tenant_operation_failures_total{component="tenant_worker",operation="process_kicked",result="panic",tenant_id="` + panicTenant + `",tidbcloud_org_id="org-panic-slot"} `
	if !strings.Contains(text, want) {
		t.Fatalf("processKicked recover must record panic failure series:\nwant %q\ngot:\n%s", want, text)
	}
}

func TestPanicResultClassifiedAsFailure(t *testing.T) {
	// Pin process_kicked panic on the tenant failure series (cardinality:
	// tenant_id only on failure results). Companion assertion lives in
	// pkg/metrics.TestTenantOperationFailuresUseSeparateFailureOnlyMetric.
	const tenantID = "tenant-panic-class"
	const orgID = "org-panic-class"
	metrics.RecordTenantOperationCountWithOrg(tenantID, orgID, "tenant_worker", "process_kicked", "panic")
	metrics.RecordTenantOperationCountWithOrg(tenantID, orgID, "tenant_worker", "slot", "busy") // non-failure

	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	wantFail := `drive9_tenant_operation_failures_total{component="tenant_worker",operation="process_kicked",result="panic",tenant_id="` + tenantID + `",tidbcloud_org_id="` + orgID + `"} `
	if !strings.Contains(text, wantFail) {
		t.Fatalf("panic must attribute on tenant failure series:\nwant substring %q\ngot:\n%s", wantFail, text)
	}
	if strings.Contains(text, `drive9_tenant_operation_failures_total{component="tenant_worker",operation="slot",result="busy"`) {
		t.Fatalf("non-failure result busy must not create tenant failure series:\n%s", text)
	}
	if !strings.Contains(text, `drive9_service_operations_total{component="tenant_worker",operation="process_kicked",result="panic"}`) {
		t.Fatalf("panic must also appear on aggregate operations series:\n%s", text)
	}
}

func TestIsContextDoneErrWrapped(t *testing.T) {
	t.Parallel()
	if isContextDoneErr(nil) {
		t.Fatal("nil must not be context-done")
	}
	if !isContextDoneErr(context.Canceled) {
		t.Fatal("plain context.Canceled")
	}
	if !isContextDoneErr(context.DeadlineExceeded) {
		t.Fatal("plain context.DeadlineExceeded")
	}
	// Wrapped form matches process+retry and DB layers (file_gc_worker.go).
	if !isContextDoneErr(fmt.Errorf("process file gc task x: %w; update retry: boom", context.Canceled)) {
		t.Fatal("wrapped context.Canceled must suppress drain failure metrics")
	}
	if !isContextDoneErr(fmt.Errorf("claim: %w", context.DeadlineExceeded)) {
		t.Fatal("wrapped context.DeadlineExceeded must suppress drain failure metrics")
	}
	if isContextDoneErr(fmt.Errorf("some other error")) {
		t.Fatal("unrelated error must not be treated as context-done")
	}
}

func TestDrainFileGCStopEmptyWithBackoffOnly(t *testing.T) {
	// Pins drain call-site claim-parity: a future available_at row must not
	// look like empty_with_backlog (CountClaimable vs CountQueued regression).
	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC().Truncate(time.Second)
	if _, err := store.EnqueueFileGCTaskTx(store.DB(), &datastore.FileGCTask{
		TaskID:      "gc-backoff-only",
		FileID:      "file-backoff-only",
		StorageType: datastore.StorageDB9,
		StorageRef:  "db9://backoff",
		Status:      datastore.FileGCTaskQueued,
		AvailableAt: now.Add(time.Hour),
		CreatedAt:   now,
		UpdatedAt:   now,
	}); err != nil {
		t.Fatal(err)
	}
	queued, err := store.CountQueuedFileGCTasks(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if queued != 1 {
		t.Fatalf("queued=%d, want 1", queued)
	}

	m := &tenantWorkerManager{
		lastRateLimitedLog: make(map[string]time.Time),
	}
	target := &tenantTarget{
		tenantID:       "tenant-drain-backoff",
		tidbCloudOrgID: "org-drain-backoff",
		backend:        b,
		store:          store,
		release:        func() {},
	}
	hitCap := m.drainFileGC(context.Background(), target)
	if hitCap {
		t.Fatal("backoff-only queue must not hit drain cap")
	}

	rec := httptest.NewRecorder()
	// Metrics are process-global. A failure from another tenant must not make
	// this drain look like it emitted stop_error.
	metrics.RecordTenantOperationCountWithOrg("foreign-drain-failure", "foreign-org", "file_gc", "drain", "stop_error")
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_service_operations_total{component="file_gc",operation="drain",result="stop_empty"}`) {
		t.Fatalf("expected stop_empty for in-backoff-only queue:\n%s", text)
	}
	if strings.Contains(text, `drive9_service_operations_total{component="file_gc",operation="drain",result="empty_with_backlog"}`) {
		t.Fatalf("in-backoff rows must not emit empty_with_backlog:\n%s", text)
	}
	if strings.Contains(text, `drive9_tenant_operation_failures_total{component="file_gc",operation="drain",result="stop_error",tenant_id="tenant-drain-backoff"`) {
		t.Fatalf("empty queue must not emit stop_error failure series:\n%s", text)
	}
}

func TestDrainFileGCCanceledContextNoFailureMetric(t *testing.T) {
	// Canceled ctx exits drain before process; must not pollute failure series.
	// Companion: isContextDoneErr wraps cancel so ProcessOne cancel errors also
	// skip stop_error (see TestIsContextDoneErrWrapped).
	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}
	m := &tenantWorkerManager{lastRateLimitedLog: make(map[string]time.Time)}
	target := &tenantTarget{
		tenantID:       "tenant-drain-cancel",
		tidbCloudOrgID: "org-drain-cancel",
		backend:        b,
		store:          store,
		release:        func() {},
	}
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	if m.drainFileGC(ctx, target) {
		t.Fatal("canceled drain must not hit cap")
	}
	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	// Tenant-attributed failure series for this tenant must stay absent.
	if strings.Contains(text, `tenant_id="tenant-drain-cancel"`) &&
		strings.Contains(text, `drive9_tenant_operation_failures_total`) &&
		(strings.Contains(text, `result="stop_error"`) || strings.Contains(text, `result="stop_empty_count_error"`)) {
		t.Fatalf("canceled drain must not emit tenant failure series:\n%s", text)
	}
}

func TestDrainFileGCProcessesDueTaskStopEmpty(t *testing.T) {
	// Due StorageDB9 task (no central meta) processes successfully then stop_empty.
	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC().Truncate(time.Second)
	if _, err := store.EnqueueFileGCTaskTx(store.DB(), &datastore.FileGCTask{
		TaskID:      "gc-due-db9",
		FileID:      "file-due-db9",
		StorageType: datastore.StorageDB9,
		StorageRef:  "db9://due",
		Status:      datastore.FileGCTaskQueued,
		AvailableAt: now.Add(-time.Second),
		CreatedAt:   now,
		UpdatedAt:   now,
	}); err != nil {
		t.Fatal(err)
	}
	m := &tenantWorkerManager{lastRateLimitedLog: make(map[string]time.Time)}
	target := &tenantTarget{
		tenantID:       "tenant-drain-due",
		tidbCloudOrgID: "org-drain-due",
		backend:        b,
		store:          store,
		release:        func() {},
	}
	if m.drainFileGC(context.Background(), target) {
		t.Fatal("single due task must not hit batch cap")
	}
	got, err := store.GetFileGCTaskByFileID(context.Background(), "file-due-db9")
	if err != nil {
		t.Fatal(err)
	}
	if got.Status != datastore.FileGCTaskSucceeded {
		t.Fatalf("task status=%s, want succeeded", got.Status)
	}
	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_service_operations_total{component="file_gc",operation="drain",result="stop_empty"}`) {
		t.Fatalf("after processing due task expect stop_empty:\n%s", text)
	}
}

func TestDrainFileGCContinuesAfterProcessError(t *testing.T) {
	// Regression for PR #56 review: ProcessOne returns (true, err) after claim +
	// process failure + successful RetryFileGCTask. drainFileGC must continue
	// so a later due healthy task is still processed in the same kick
	// (main behavior; stop-on-any-error strands N+1..batch until external kick).
	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}
	// No metaStore / object-GC enqueue on this backend: StorageS3 process fails
	// with "object gc candidate enqueue is not configured", then retries.
	now := time.Now().UTC().Truncate(time.Second)
	failTask := &datastore.FileGCTask{
		TaskID:      "gc-fail-s3",
		FileID:      "file-fail-s3",
		StorageType: datastore.StorageS3,
		StorageRef:  "s3://bucket/fail-blob",
		Status:      datastore.FileGCTaskQueued,
		AvailableAt: now.Add(-2 * time.Second),
		CreatedAt:   now,
		UpdatedAt:   now,
	}
	okTask := &datastore.FileGCTask{
		TaskID:      "gc-ok-db9",
		FileID:      "file-ok-db9",
		StorageType: datastore.StorageDB9,
		StorageRef:  "db9://ok",
		Status:      datastore.FileGCTaskQueued,
		AvailableAt: now.Add(-time.Second),
		CreatedAt:   now.Add(time.Millisecond),
		UpdatedAt:   now,
	}
	for _, task := range []*datastore.FileGCTask{failTask, okTask} {
		if _, err := store.EnqueueFileGCTaskTx(store.DB(), task); err != nil {
			t.Fatal(err)
		}
	}

	m := &tenantWorkerManager{lastRateLimitedLog: make(map[string]time.Time)}
	target := &tenantTarget{
		tenantID:       "tenant-drain-continue",
		tidbCloudOrgID: "org-drain-continue",
		backend:        b,
		store:          store,
		release:        func() {},
	}
	if m.drainFileGC(context.Background(), target) {
		t.Fatal("two-task queue must not hit batch cap")
	}

	failed, err := store.GetFileGCTaskByFileID(context.Background(), "file-fail-s3")
	if err != nil {
		t.Fatal(err)
	}
	if failed.Status != datastore.FileGCTaskQueued {
		t.Fatalf("failed task status=%s, want queued (retry scheduled)", failed.Status)
	}
	if failed.AttemptCount < 1 {
		t.Fatalf("failed task attempt_count=%d, want >= 1", failed.AttemptCount)
	}
	// available_at must be in the future so it does not re-claim in the same drain.
	if !failed.AvailableAt.After(now) {
		t.Fatalf("retried task available_at=%v should be after seed now=%v", failed.AvailableAt, now)
	}

	ok, err := store.GetFileGCTaskByFileID(context.Background(), "file-ok-db9")
	if err != nil {
		t.Fatal(err)
	}
	if ok.Status != datastore.FileGCTaskSucceeded {
		t.Fatalf("healthy task after process-error must still succeed in same drain; status=%s", ok.Status)
	}

	rec := httptest.NewRecorder()
	// Metrics are process-global. Scope the negative assertion to this tenant
	// so unrelated drains cannot make the test order-dependent.
	metrics.RecordTenantOperationCountWithOrg("foreign-drain-failure", "foreign-org", "file_gc", "drain", "stop_error")
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_service_operations_total{component="file_gc",operation="drain",result="task_error"}`) {
		t.Fatalf("expected task_error (continue path), not stop_error:\n%s", text)
	}
	if strings.Contains(text, `drive9_tenant_operation_failures_total{component="file_gc",operation="drain",result="stop_error",tenant_id="tenant-drain-continue"`) {
		t.Fatalf("process-after-claim error must not stop the batch with stop_error:\n%s", text)
	}
	if !strings.Contains(text, `drive9_service_operations_total{component="file_gc",operation="drain",result="stop_empty"}`) {
		t.Fatalf("after continue+process healthy task expect stop_empty:\n%s", text)
	}
}

func TestWorkerLoopPanicWaitsPollInterval(t *testing.T) {
	// Regression (qiffang): after pollFallbackOnce panics, workerLoop must wait
	// for PollInterval instead of tight-looping while shouldPollFallback stays true.
	// No t.Parallel: mutates package-global pollFallbackOnceTestHook.
	//
	// Discriminating assertions (must fail under the pre-fix tight loop):
	//  1. inter-panic span for the first panicN calls is ~ (N-1)*PollInterval
	//  2. total panic count over a fixed window is O(window/PollInterval), not unbounded
	const panicN = 3
	const pollEvery = 50 * time.Millisecond
	var (
		mu         sync.Mutex
		panicAt    []time.Time
		panicCalls int
		target     *tenantWorkerManager
	)

	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}
	m := newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{
		PollInterval: pollEvery,
		Workers:      1,
	}, 0)
	if m == nil {
		t.Fatal("manager")
	}
	target = m
	// Scope the hook to this manager only: other package tests running under
	// t.Parallel may also call pollFallbackOnce and must not trip this counter.
	pollFallbackOnceTestHook = func(got *tenantWorkerManager) {
		if got != target {
			return
		}
		mu.Lock()
		panicCalls++
		panicAt = append(panicAt, time.Now())
		mu.Unlock()
		// Always panic: a tight loop would produce thousands of samples in the
		// observation window; the rate-bounded path stays O(window/PollInterval).
		panic("test pollFallback panic")
	}
	t.Cleanup(func() { pollFallbackOnceTestHook = nil })

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	m.Start(ctx)
	// Observe long enough for several rate-limited panics.
	window := time.Duration(panicN+3) * pollEvery
	time.Sleep(window)
	cancel()
	m.Stop()

	mu.Lock()
	n := panicCalls
	times := append([]time.Time(nil), panicAt...)
	mu.Unlock()

	if n < panicN {
		t.Fatalf("expected at least %d panics from fallback poll, got %d", panicN, n)
	}
	// Upper bound: with post-panic fresh-timer backoff, count is ~window/PollInterval.
	// Tight loop would be orders of magnitude higher within the same wall window.
	// Generous slack absorbs timer resolution / CI scheduling without admitting a spin.
	maxOK := int(window/pollEvery) + 6
	if n > maxOK {
		t.Fatalf("panic count %d exceeds rate-bounded max %d over %v; tight-loop suspected", n, maxOK, window)
	}
	// Inter-panic spacing for the first panicN samples.
	span := times[panicN-1].Sub(times[0])
	minSpan := time.Duration(panicN-1) * (pollEvery * 3 / 4)
	if span < minSpan {
		t.Fatalf("first %d panics spanned %v, want >= %v (rate-bounded by PollInterval)", panicN, span, minSpan)
	}
}

func TestWorkerLoopPanicBackoffStartsAfterRecovery(t *testing.T) {
	// Regression (qiffang B3): when the panicking poll outlasts PollInterval,
	// the post-panic backoff must still wait a full PollInterval measured from
	// recovery — a tick buffered during the slow work must not satisfy it.
	// No t.Parallel: mutates package-global pollFallbackOnceTestHook.
	const pollEvery = 50 * time.Millisecond
	const slowWork = 3 * pollEvery // > PollInterval: stale tick guaranteed pre-fix
	var (
		mu     sync.Mutex
		calls  []time.Time
		target *tenantWorkerManager
	)
	secondCall := make(chan struct{})

	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}
	m := newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{
		PollInterval: pollEvery,
		Workers:      1,
	}, 0)
	if m == nil {
		t.Fatal("manager")
	}
	target = m
	pollFallbackOnceTestHook = func(got *tenantWorkerManager) {
		if got != target {
			return
		}
		mu.Lock()
		calls = append(calls, time.Now())
		n := len(calls)
		mu.Unlock()
		switch n {
		case 1:
			time.Sleep(slowWork) // outlast PollInterval before panicking
		case 2:
			close(secondCall)
		}
		panic("test pollFallback slow panic")
	}
	t.Cleanup(func() { pollFallbackOnceTestHook = nil })

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	m.Start(ctx)
	select {
	case <-secondCall:
	case <-time.After(slowWork + 10*pollEvery + 5*time.Second):
		cancel()
		m.Stop()
		t.Fatal("second fallback poll never happened; worker wedged")
	}
	cancel()
	m.Stop()

	mu.Lock()
	got := append([]time.Time(nil), calls...)
	mu.Unlock()
	if len(got) < 2 {
		t.Fatalf("calls = %d, want >= 2", len(got))
	}
	// Discriminating: entry-to-entry gap = slowWork + backoff. Pre-fix the
	// backoff collapses to ~0 (buffered tick); post-fix it is >= PollInterval.
	gap := got[1].Sub(got[0])
	if min := slowWork + pollEvery*3/4; gap < min {
		t.Fatalf("gap between panicking polls = %v, want >= %v (backoff must start at recovery, not reuse a stale tick)", gap, min)
	}
}

func withObservabilityLogger(t *testing.T, level zapcore.Level) *observer.ObservedLogs {
	t.Helper()
	core, recorded := observer.New(level)
	prev := logger.L()
	logger.Set(zap.New(core))
	t.Cleanup(func() { logger.Set(prev) })
	return recorded
}

func newBareKickManager(kicks chan kickMsg) *tenantWorkerManager {
	m := &tenantWorkerManager{
		inflight:           make(map[string]int),
		kickPending:        make(map[string]pendingKick),
		kicks:              kicks,
		lastRateLimitedLog: make(map[string]time.Time),
		opts:               TenantWorkerOptions{PerTenantConcurrency: 1},
	}
	m.opts.normalize()
	return m
}

func TestKickWithOrgFileGCCoalescedLogsTenantEvidence(t *testing.T) {
	// B2: WorkFileGC coalesce before processKicked must emit a default-visible
	// (Info) tenant-attributed log, rate-limited so a burst does not flood.
	// No t.Parallel: swaps global logger.
	recorded := withObservabilityLogger(t, zap.InfoLevel)
	m := newBareKickManager(make(chan kickMsg, tenantKickQueueCapacity))
	const tenantID = "tenant-kick-coalesce-obs"
	m.KickWithOrg(tenantID, "org-coalesce", WorkFileGC) // queued
	m.KickWithOrg(tenantID, "org-coalesce", WorkFileGC) // coalesced → log
	m.KickWithOrg(tenantID, "org-coalesce", WorkFileGC) // coalesced → rate-limited

	entries := recorded.FilterMessage("tenant_worker_kick_coalesced").All()
	if len(entries) != 1 {
		t.Fatalf("coalesced log entries = %d, want 1 (rate-limited, no flood)", len(entries))
	}
	fields := entries[0].ContextMap()
	if fields["tenant_id"] != tenantID {
		t.Fatalf("tenant_id = %v, want %s", fields["tenant_id"], tenantID)
	}
	pendingMask, ok := fields["pending_mask"].(int64)
	if !ok {
		// zap may encode as int
		if pm, ok2 := fields["pending_mask"].(int); ok2 {
			pendingMask = int64(pm)
		} else {
			t.Fatalf("pending_mask type %T = %v", fields["pending_mask"], fields["pending_mask"])
		}
	}
	if int(pendingMask)&WorkFileGC == 0 {
		t.Fatalf("pending_mask=%v missing WorkFileGC", fields["pending_mask"])
	}

	// Cardinality: coalesced must not create tenant failure series.
	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	if strings.Contains(text, `drive9_tenant_operation_failures_total`) &&
		strings.Contains(text, `tenant_id="`+tenantID+`"`) {
		t.Fatalf("coalesce must not attribute tenant on failure series:\n%s", text)
	}
}

func TestKickWithOrgFileGCDelayedLogsTenantEvidence(t *testing.T) {
	// B2: channel-full WorkFileGC must emit rate-limited Warn with tenant_id.
	// No t.Parallel: swaps global logger.
	recorded := withObservabilityLogger(t, zap.InfoLevel) // Warn is visible at Info core
	// Capacity 1: filler occupies the channel; target tenant hits delayed path.
	m := newBareKickManager(make(chan kickMsg, 1))
	m.KickWithOrg("filler-tenant", "org-f", WorkFileGC) // fills channel
	const tenantID = "tenant-kick-delay-obs"
	m.KickWithOrg(tenantID, "org-delay", WorkFileGC) // delayed → log
	// Clear pending so the next KickWithOrg can hit delayed again (not coalesce).
	// Channel still full (filler unconsumed).
	_ = m.takePendingKick(tenantID)
	m.KickWithOrg(tenantID, "org-delay", WorkFileGC) // delayed again → rate-limited

	entries := recorded.FilterMessage("tenant_worker_kick_delayed").All()
	if len(entries) != 1 {
		t.Fatalf("delayed log entries = %d, want 1 (rate-limited)", len(entries))
	}
	if fields := entries[0].ContextMap(); fields["tenant_id"] != tenantID {
		t.Fatalf("tenant_id = %v, want %s", fields["tenant_id"], tenantID)
	}
	if entries[0].Level != zap.WarnLevel {
		t.Fatalf("delayed log level = %v, want warn", entries[0].Level)
	}
}

func TestProcessKickedSlotBusyRecordsFileGCSchedError(t *testing.T) {
	// FileGC kick that loses the tenant slot must hit the tenant failure series
	// (slot_busy_error) so lagging GC can be paged without waiting for drain.
	// kickRef(local) with meta/pool nil accepts WorkFileGC without fallback.
	m := newBareKickManager(make(chan kickMsg, 8))
	if !m.tryClaimTenantSlot(tenantLocalID) {
		t.Fatal("pre-claim local slot to force busy")
	}
	m.processKicked(context.Background(), tenantLocalID, "org-slot-busy", WorkFileGC)

	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	want := `drive9_tenant_operation_failures_total{component="file_gc",operation="sched",result="slot_busy_error",tenant_id="` + tenantLocalID + `",tidbcloud_org_id="org-slot-busy"} `
	if !strings.Contains(text, want) {
		t.Fatalf("missing file_gc sched slot_busy_error failure series:\nwant %q\ngot:\n%s", want, text)
	}
	m.releaseTenantSlot(tenantLocalID)
}

func TestProcessKickedScanSlotBusyPreservesCoalescedWork(t *testing.T) {
	m := newBareKickManager(make(chan kickMsg, 8))
	m.KickWithOrg(tenantLocalID, "org-scan", WorkStorageTierScan)
	m.KickWithOrg(tenantLocalID, "org-scan", WorkFileGC)
	if !m.tryClaimTenantSlot(tenantLocalID) {
		t.Fatal("pre-claim local slot to force busy")
	}
	m.processKicked(context.Background(), tenantLocalID, "org-scan", WorkStorageTierScan)
	m.releaseTenantSlot(tenantLocalID)

	if got := m.peekPendingWorkMask(tenantLocalID); got&WorkFileGC == 0 {
		t.Fatalf("coalesced work mask = %d, want WorkFileGC preserved", got)
	}
}

func TestDrainFileGCExportsClaimableGauges(t *testing.T) {
	// Worker-path wiring: claimable backlog must appear on sparse gauges after
	// drain observation, and disappear when the queue is empty (alert contract).
	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC().Truncate(time.Second)
	// Two due tasks; drain processes them then stop_empty → gauges deleted.
	for i, id := range []string{"gauge-gc-1", "gauge-gc-2"} {
		if _, err := store.EnqueueFileGCTaskTx(store.DB(), &datastore.FileGCTask{
			TaskID:      id,
			FileID:      id,
			StorageType: datastore.StorageDB9,
			StorageRef:  "db9://" + id,
			Status:      datastore.FileGCTaskQueued,
			AvailableAt: now.Add(-time.Duration(i+1) * time.Second),
			CreatedAt:   now,
			UpdatedAt:   now,
		}); err != nil {
			t.Fatal(err)
		}
	}
	m := &tenantWorkerManager{lastRateLimitedLog: make(map[string]time.Time)}
	const tenantID = "tenant-gauge-export"
	const orgID = "org-gauge-export"
	target := &tenantTarget{
		tenantID:       tenantID,
		tidbCloudOrgID: orgID,
		backend:        b,
		store:          store,
		release:        func() {},
	}
	// Direct observe still exports COUNT-based depth; processKicked no longer
	// does this before Claim (Claim-path drain defer is the alert oracle).
	m.observeFileGCClaimableBacklog(context.Background(), target)
	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_file_gc_claimable{tenant_id="`+tenantID+`",tidbcloud_org_id="`+orgID+`"} 2`) {
		t.Fatalf("pre-observe missing claimable=2 gauge:\n%s", text)
	}
	if !strings.Contains(text, `drive9_file_gc_oldest_claimable_age_seconds{tenant_id="`+tenantID+`",tidbcloud_org_id="`+orgID+`"}`) {
		t.Fatalf("pre-observe missing oldest age gauge:\n%s", text)
	}

	if m.drainFileGC(context.Background(), target) {
		t.Fatal("two tasks must not hit batch cap")
	}
	rec = httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text = rec.Body.String()
	if strings.Contains(text, `drive9_file_gc_claimable{tenant_id="`+tenantID+`"`) {
		t.Fatalf("after drain to empty, claimable gauge must be deleted:\n%s", text)
	}
	if strings.Contains(text, `drive9_file_gc_oldest_claimable_age_seconds{tenant_id="`+tenantID+`"`) {
		t.Fatalf("after drain to empty, oldest age gauge must be deleted:\n%s", text)
	}
}

func TestDrainFileGCClaimEmptyWithBacklogClearsGaugeAndReKicks(t *testing.T) {
	// Claim empty + COUNT>0 must not export the stale-backlog gauge (that
	// leftover COUNT is what paged phantom tenants). It still records
	// empty_with_backlog and schedules a delayed FileGC re-kick.
	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}

	origBase := fileGCMismatchReKickBase
	origMax := fileGCMismatchReKickMax
	fileGCMismatchReKickBase = time.Millisecond
	fileGCMismatchReKickMax = time.Millisecond
	t.Cleanup(func() {
		fileGCMismatchReKickBase = origBase
		fileGCMismatchReKickMax = origMax
		drainFileGCClaimableAfterEmptyTestHook = nil
	})
	drainFileGCClaimableAfterEmptyTestHook = func() (int64, time.Time, error) {
		return 5, time.Now().UTC().Add(-2 * time.Hour), nil
	}

	kicks := make(chan kickMsg, 1)
	m := newBareKickManager(kicks)
	const tenantID = "tenant-drain-mismatch"
	const orgID = "org-drain-mismatch"
	target := &tenantTarget{
		tenantID:       tenantID,
		tidbCloudOrgID: orgID,
		backend:        b,
		store:          store,
		release:        func() {},
	}
	if m.drainFileGC(context.Background(), target) {
		t.Fatal("empty claim must not hit drain cap")
	}

	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_service_operations_total{component="file_gc",operation="drain",result="empty_with_backlog"}`) {
		t.Fatalf("expected empty_with_backlog counter:\n%s", text)
	}
	if strings.Contains(text, `drive9_file_gc_claimable{tenant_id="`+tenantID+`"`) {
		t.Fatalf("mismatch must not export claimable gauge:\n%s", text)
	}
	if strings.Contains(text, `drive9_file_gc_drain_hit_cap{tenant_id="`+tenantID+`"`) {
		t.Fatalf("mismatch must not export drain_hit_cap gauge:\n%s", text)
	}

	select {
	case msg := <-kicks:
		if msg.tenantID != tenantID {
			t.Fatalf("rekick tenant=%s, want %s", msg.tenantID, tenantID)
		}
		if msg.workMask&WorkFileGC == 0 {
			t.Fatalf("rekick work_mask=%d, want WorkFileGC", msg.workMask)
		}
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for mismatch FileGC re-kick")
	}
}

func TestDrainFileGCCountErrorSchedulesReKick(t *testing.T) {
	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}
	origBase := fileGCMismatchReKickBase
	origMax := fileGCMismatchReKickMax
	fileGCMismatchReKickBase = time.Millisecond
	fileGCMismatchReKickMax = time.Millisecond
	t.Cleanup(func() {
		fileGCMismatchReKickBase = origBase
		fileGCMismatchReKickMax = origMax
		drainFileGCClaimableAfterEmptyTestHook = nil
	})
	drainFileGCClaimableAfterEmptyTestHook = func() (int64, time.Time, error) {
		return 0, time.Time{}, errors.New("count failed")
	}
	kicks := make(chan kickMsg, 1)
	m := newBareKickManager(kicks)
	const tenantID = "tenant-drain-count-error"
	target := &tenantTarget{
		tenantID:       tenantID,
		tidbCloudOrgID: "org-drain-count-error",
		backend:        b,
		store:          store,
		release:        func() {},
	}
	if m.drainFileGC(context.Background(), target) {
		t.Fatal("count-error empty claim must not hit drain cap")
	}
	if strings.Contains(prometheusText(t), `drive9_file_gc_claimable{tenant_id="`+tenantID+`"`) {
		t.Fatal("count-error must not export COUNT-based claimable")
	}
	select {
	case msg := <-kicks:
		if msg.tenantID != tenantID || msg.workMask&WorkFileGC == 0 {
			t.Fatalf("count-error rekick tenant=%s mask=%d", msg.tenantID, msg.workMask)
		}
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for count-error FileGC re-kick")
	}
}

func TestDrainFileGCMismatchReKickMaxAttempts(t *testing.T) {
	// After fileGCMismatchReKickMaxAttempts delayed kicks, further mismatches
	// must not schedule another FileGC kick (no tight loop).
	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}
	origBase := fileGCMismatchReKickBase
	origMax := fileGCMismatchReKickMax
	fileGCMismatchReKickBase = time.Millisecond
	fileGCMismatchReKickMax = time.Millisecond
	t.Cleanup(func() {
		fileGCMismatchReKickBase = origBase
		fileGCMismatchReKickMax = origMax
		drainFileGCClaimableAfterEmptyTestHook = nil
	})
	drainFileGCClaimableAfterEmptyTestHook = func() (int64, time.Time, error) {
		return 3, time.Now().UTC().Add(-time.Hour), nil
	}

	kicks := make(chan kickMsg, 8)
	m := newBareKickManager(kicks)
	const tenantID = "tenant-mismatch-cap"
	target := &tenantTarget{
		tenantID:       tenantID,
		tidbCloudOrgID: "org-mismatch-cap",
		backend:        b,
		store:          store,
		release:        func() {},
	}
	waitKick := func() {
		t.Helper()
		select {
		case msg := <-kicks:
			if msg.tenantID != tenantID || msg.workMask&WorkFileGC == 0 {
				t.Fatalf("unexpected kick tenant=%s mask=%d", msg.tenantID, msg.workMask)
			}
		case <-time.After(time.Second):
			t.Fatal("timed out waiting for mismatch re-kick")
		}
		_ = m.takePendingKick(tenantID)
	}
	for i := 0; i < fileGCMismatchReKickMaxAttempts; i++ {
		if m.drainFileGC(context.Background(), target) {
			t.Fatal("mismatch drain must not hit cap")
		}
		waitKick()
	}
	if m.drainFileGC(context.Background(), target) {
		t.Fatal("exhausted mismatch drain must not hit cap")
	}
	select {
	case msg := <-kicks:
		t.Fatalf("5th mismatch must not re-kick, got tenant=%s mask=%d", msg.tenantID, msg.workMask)
	case <-time.After(50 * time.Millisecond):
	}
	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	if strings.Contains(text, `drive9_file_gc_claimable{tenant_id="`+tenantID+`"`) {
		t.Fatalf("exhausted mismatch must not export claimable:\n%s", text)
	}
	if !strings.Contains(text, `drive9_file_gc_claim_stalled{tenant_id="`+tenantID+`",tidbcloud_org_id="org-mismatch-cap"} 1`) {
		t.Fatalf("exhausted mismatch must export claim_stalled=1:\n%s", text)
	}
}

func TestDrainFileGCHitCapExportsDrainHitCapGauge(t *testing.T) {
	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC().Truncate(time.Second)
	for i := 0; i < fileGCDrainBatchSize+1; i++ {
		id := fmt.Sprintf("gc-cap-%03d", i)
		if _, err := store.EnqueueFileGCTaskTx(store.DB(), &datastore.FileGCTask{
			TaskID:      id,
			FileID:      id,
			StorageType: datastore.StorageDB9,
			StorageRef:  "db9://" + id,
			Status:      datastore.FileGCTaskQueued,
			AvailableAt: now.Add(-time.Duration(i+1) * time.Millisecond),
			CreatedAt:   now,
			UpdatedAt:   now,
		}); err != nil {
			t.Fatal(err)
		}
	}
	m := &tenantWorkerManager{lastRateLimitedLog: make(map[string]time.Time)}
	const tenantID = "tenant-drain-hit-cap"
	const orgID = "org-drain-hit-cap"
	target := &tenantTarget{
		tenantID:       tenantID,
		tidbCloudOrgID: orgID,
		backend:        b,
		store:          store,
		release:        func() {},
	}
	if !m.drainFileGC(context.Background(), target) {
		t.Fatal("batch+1 tasks must hit drain cap")
	}
	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	text := rec.Body.String()
	if !strings.Contains(text, `drive9_service_operations_total{component="file_gc",operation="drain",result="hit_cap"}`) {
		t.Fatalf("expected hit_cap counter:\n%s", text)
	}
	if !strings.Contains(text, `drive9_file_gc_drain_hit_cap{tenant_id="`+tenantID+`",tidbcloud_org_id="`+orgID+`"} 1`) {
		t.Fatalf("expected drain_hit_cap=1:\n%s", text)
	}
	if !strings.Contains(text, `drive9_file_gc_claimable{tenant_id="`+tenantID+`",tidbcloud_org_id="`+orgID+`"}`) {
		t.Fatalf("hit_cap drain must still export remaining claimable:\n%s", text)
	}
}

func TestFileCountGaugeDropsOnOwnershipLossAndNonOwnerCannotRepublish(t *testing.T) {
	// Ownership loss must delete this process's file-count series even while
	// request-path metrics keep the tenant's series state fresh (retention
	// alone never expires it). A still-owned tenant keeps its series, and a
	// non-owner or in-flight observation must not recreate the dropped ones.
	const lostID = "tenant-filecount-owner-lost"
	const keepID = "tenant-filecount-owner-keep"
	const orgID = "org-filecount-owner"
	t.Cleanup(func() {
		metrics.DeleteTenantCounters(lostID)
		metrics.DeleteTenantCounters(keepID)
	})
	dbFiles, s3Files := int64(2), int64(1)
	metrics.ReplaceTenantFileCountSnapshotWithOrg(lostID, orgID, 3, &dbFiles, &s3Files, true)
	metrics.ReplaceTenantFileCountSnapshotWithOrg(keepID, orgID, 3, &dbFiles, &s3Files, true)

	owned := map[string]bool{lostID: true, keepID: true}
	m := &tenantWorkerManager{
		ownsTenant: func(id string) bool { return owned[id] },
	}
	m.dropUnownedFileCountGauges()
	if got := prometheusText(t); !strings.Contains(got, `drive9_tenant_file_count{state="confirmed",storage_type="db9",tenant_id="`+lostID+`"`) {
		t.Fatalf("owner drop must not clear still-owned lost tenant:\n%s", got)
	}

	owned[lostID] = false
	// A request-path metric on the old pod keeps the tenant warm, proving the
	// ownership sweep — not retention — is what removes the stale gauge.
	metrics.RecordTenantRequestCountWithOrg(lostID, orgID, "fs", "read", 200)
	m.dropUnownedFileCountGauges()
	text := prometheusText(t)
	for _, line := range strings.Split(text, "\n") {
		if strings.HasPrefix(line, "drive9_tenant_file_count{") && strings.Contains(line, `tenant_id="`+lostID+`"`) {
			t.Fatalf("unowned tenant still exports file count gauge: %s", line)
		}
	}
	if !strings.Contains(text, `drive9_tenant_file_count{state="confirmed",storage_type="db9",tenant_id="`+keepID+`"`) {
		t.Fatalf("still-owned tenant lost file count series:\n%s", text)
	}

	// A queued/in-flight observation landing after the ring move must clear
	// instead of publish.
	m.withOwnedFileCountGauge(lostID, func() {
		metrics.ReplaceTenantFileCountSnapshotWithOrg(lostID, orgID, 5, nil, nil, false)
	})
	text = prometheusText(t)
	for _, line := range strings.Split(text, "\n") {
		if strings.HasPrefix(line, "drive9_tenant_file_count{") && strings.Contains(line, `tenant_id="`+lostID+`"`) {
			t.Fatalf("non-owner publish recreated leftover file count series: %s", line)
		}
	}

	// Ring moves mid-publish: the post-publish recheck drops the just-written
	// series.
	m.withOwnedFileCountGauge(keepID, func() {
		metrics.ReplaceTenantFileCountSnapshotWithOrg(keepID, orgID, 7, nil, nil, false)
		owned[keepID] = false
	})
	text = prometheusText(t)
	for _, line := range strings.Split(text, "\n") {
		if strings.HasPrefix(line, "drive9_tenant_file_count{") && strings.Contains(line, `tenant_id="`+keepID+`"`) {
			t.Fatalf("series published during ownership loss survived the post-publish recheck: %s", line)
		}
	}
}

func TestFileGCGaugesDropOnOwnershipLossWithoutTouchingOwnedDrain(t *testing.T) {
	// Ownership loss on a still-cached tenant must delete this process's
	// FileGC series without closeEntry/ForgetTenant. A still-owned tenant
	// with an active drain keeps its series.
	const lostID = "tenant-filegc-owner-lost"
	const keepID = "tenant-filegc-owner-keep"
	const orgID = "org-filegc-owner"
	metrics.ObserveFileGCClaimableBacklog(lostID, orgID, 4, time.Hour)
	metrics.ObserveFileGCDrainHitCap(lostID, orgID, true)
	metrics.ObserveFileGCClaimStalled(lostID, orgID, true)
	metrics.ObserveFileGCClaimableBacklog(keepID, orgID, 9, 2*time.Hour)
	metrics.ObserveFileGCDrainHitCap(keepID, orgID, true)

	owned := map[string]bool{lostID: true, keepID: true}
	m := &tenantWorkerManager{
		ownsTenant:             func(id string) bool { return owned[id] },
		lastRateLimitedLog:     make(map[string]time.Time),
		fileGCMismatchTimers:   make(map[string]*time.Timer),
		fileGCMismatchAttempts: make(map[string]int),
	}
	m.dropUnownedFileGCGauges()
	if got := prometheusText(t); !strings.Contains(got, `drive9_file_gc_claimable{tenant_id="`+lostID+`"`) {
		t.Fatalf("owner drop must not clear still-owned lost tenant:\n%s", got)
	}

	owned[lostID] = false
	m.dropUnownedFileGCGauges()
	text := prometheusText(t)
	if strings.Contains(text, `drive9_file_gc_claimable{tenant_id="`+lostID+`"`) ||
		strings.Contains(text, `drive9_file_gc_drain_hit_cap{tenant_id="`+lostID+`"`) ||
		strings.Contains(text, `drive9_file_gc_claim_stalled{tenant_id="`+lostID+`"`) ||
		strings.Contains(text, `drive9_file_gc_oldest_claimable_age_seconds{tenant_id="`+lostID+`"`) {
		t.Fatalf("unowned cached tenant still exports FileGC gauges:\n%s", text)
	}
	if !strings.Contains(text, `drive9_file_gc_claimable{tenant_id="`+keepID+`",tidbcloud_org_id="`+orgID+`"} 9`) {
		t.Fatalf("still-owned active drain lost claimable series:\n%s", text)
	}
	if !strings.Contains(text, `drive9_file_gc_drain_hit_cap{tenant_id="`+keepID+`",tidbcloud_org_id="`+orgID+`"} 1`) {
		t.Fatalf("still-owned active drain lost hit_cap series:\n%s", text)
	}

	m.withOwnedFileGCGauges(lostID, func() {
		metrics.ObserveFileGCClaimableBacklog(lostID, orgID, 99, time.Hour)
		metrics.ObserveFileGCDrainHitCap(lostID, orgID, true)
	})
	text = prometheusText(t)
	if strings.Contains(text, `drive9_file_gc_claimable{tenant_id="`+lostID+`"`) {
		t.Fatalf("non-owner SET recreated leftover claimable series:\n%s", text)
	}
}

func TestObserveFileGCClaimableBacklogClearsWhenNotOwner(t *testing.T) {
	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	now := time.Now().UTC().Truncate(time.Second)
	if _, err := store.EnqueueFileGCTaskTx(store.DB(), &datastore.FileGCTask{
		TaskID:      "observe-unowned-1",
		FileID:      "observe-unowned-1",
		StorageType: datastore.StorageDB9,
		StorageRef:  "db9://observe-unowned-1",
		Status:      datastore.FileGCTaskQueued,
		AvailableAt: now.Add(-time.Second),
		CreatedAt:   now,
		UpdatedAt:   now,
	}); err != nil {
		t.Fatal(err)
	}
	const tenantID = "tenant-observe-unowned"
	const orgID = "org-observe-unowned"
	metrics.ObserveFileGCClaimableBacklog(tenantID, orgID, 3, time.Hour)
	m := &tenantWorkerManager{
		ownsTenant:         func(string) bool { return false },
		lastRateLimitedLog: make(map[string]time.Time),
	}
	target := &tenantTarget{
		tenantID:       tenantID,
		tidbCloudOrgID: orgID,
		store:          store,
		release:        func() {},
	}
	m.observeFileGCClaimableBacklog(context.Background(), target)
	if strings.Contains(prometheusText(t), `drive9_file_gc_claimable{tenant_id="`+tenantID+`"`) {
		t.Fatal("non-owner observe must delete leftover claimable series")
	}
}

func TestProcessKickedFileGCSkipsAndClearsWhenNotOwner(t *testing.T) {
	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now().UTC().Truncate(time.Second)
	for i, id := range []string{"kicked-unowned-1", "kicked-unowned-2"} {
		if _, err := store.EnqueueFileGCTaskTx(store.DB(), &datastore.FileGCTask{
			TaskID:      id,
			FileID:      id,
			StorageType: datastore.StorageDB9,
			StorageRef:  "db9://" + id,
			Status:      datastore.FileGCTaskQueued,
			AvailableAt: now.Add(-time.Duration(i+1) * time.Second),
			CreatedAt:   now,
			UpdatedAt:   now,
		}); err != nil {
			t.Fatal(err)
		}
	}
	m := newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{MaintenanceInterval: time.Hour}, 0)
	m.setOwnsTenant(func(string) bool { return false })
	const orgID = "org-kicked-unowned"
	metrics.ObserveFileGCClaimableBacklog(tenantLocalID, orgID, 2, time.Hour)
	metrics.ObserveFileGCDrainHitCap(tenantLocalID, orgID, true)
	m.processKicked(context.Background(), tenantLocalID, orgID, WorkFileGC)

	text := prometheusText(t)
	if strings.Contains(text, `drive9_file_gc_claimable{tenant_id="`+tenantLocalID+`"`) ||
		strings.Contains(text, `drive9_file_gc_drain_hit_cap{tenant_id="`+tenantLocalID+`"`) {
		t.Fatalf("non-owner processKicked left FileGC gauges:\n%s", text)
	}
	queued, err := store.CountQueuedFileGCTasks(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if queued != 2 {
		t.Fatalf("non-owner processKicked drained FileGC tasks: queued=%d, want 2", queued)
	}
}

func TestApplyRingDropsUnownedFileGCGauges(t *testing.T) {
	const tenantID = "tenant-ring-drop-filegc"
	const orgID = "org-ring-drop-filegc"
	metrics.ObserveFileGCClaimableBacklog(tenantID, orgID, 5, time.Hour)
	r := newSemanticShardResolver(nil, "pod-a", 0)
	m := &tenantWorkerManager{
		ownsTenant:           r.ownsTenantFn(),
		lastRateLimitedLog:   make(map[string]time.Time),
		fileGCMismatchTimers: make(map[string]*time.Timer),
	}
	r.setOnRingRefresh(m.dropUnownedFileGCGauges)
	r.applyRing([]string{"pod-a"})
	if !strings.Contains(prometheusText(t), `drive9_file_gc_claimable{tenant_id="`+tenantID+`"`) {
		t.Fatal("single-pod ring must keep owned FileGC series")
	}
	r.applyRing([]string{"pod-b", "pod-c"})
	if strings.Contains(prometheusText(t), `drive9_file_gc_claimable{tenant_id="`+tenantID+`"`) {
		t.Fatal("ring without self must drop FileGC series")
	}
}

func prometheusText(t *testing.T) string {
	t.Helper()
	rec := httptest.NewRecorder()
	metrics.WritePrometheus(rec)
	return rec.Body.String()
}

func TestDropUnownedFileGCGaugesClearsMismatchBackoffWithoutSeries(t *testing.T) {
	origBase := fileGCMismatchReKickBase
	origMax := fileGCMismatchReKickMax
	fileGCMismatchReKickBase = time.Hour
	fileGCMismatchReKickMax = time.Hour
	t.Cleanup(func() {
		fileGCMismatchReKickBase = origBase
		fileGCMismatchReKickMax = origMax
	})
	const tenantID = "tenant-mismatch-timer-unowned"
	owned := true
	m := newBareKickManager(make(chan kickMsg, 1))
	m.setOwnsTenant(func(string) bool { return owned })
	if !m.scheduleFileGCMismatchReKick(&tenantTarget{tenantID: tenantID, tidbCloudOrgID: "org-x"}) {
		t.Fatal("expected mismatch timer armed")
	}
	owned = false
	m.dropUnownedFileGCGauges()
	m.mu.Lock()
	_, hasTimer := m.fileGCMismatchTimers[tenantID]
	_, hasAttempt := m.fileGCMismatchAttempts[tenantID]
	m.mu.Unlock()
	if hasTimer || hasAttempt {
		t.Fatal("unowned mismatch backoff must be cleared even with no exported gauges")
	}
}

func TestProcessKickedFileGCMismatchDoesNotExportClaimable(t *testing.T) {
	initWorkerReconcileSchema(t)
	store, err := datastore.Open(testDSN)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })
	b, err := backend.New(store)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { drainFileGCClaimableAfterEmptyTestHook = nil })
	drainFileGCClaimableAfterEmptyTestHook = func() (int64, time.Time, error) {
		return 5, time.Now().UTC().Add(-2 * time.Hour), nil
	}
	m := newTenantWorkerManager(b, nil, nil, nil, TenantWorkerOptions{MaintenanceInterval: time.Hour}, 0)
	m.processKicked(context.Background(), tenantLocalID, "org-kicked-mismatch", WorkFileGC)
	if strings.Contains(prometheusText(t), `drive9_file_gc_claimable{tenant_id="`+tenantLocalID+`"`) {
		t.Fatal("mismatch processKicked must not export COUNT-based claimable")
	}
}

func TestLogFileGCReKickRateLimitedInfo(t *testing.T) {
	// B2: FileGC re-kick evidence at Info (was Debug) and rate-limited.
	// No t.Parallel: swaps global logger.
	recorded := withObservabilityLogger(t, zap.InfoLevel)
	m := newBareKickManager(make(chan kickMsg, 8))
	const tenantID = "tenant-rekick-obs"
	ctx := context.Background()
	m.logFileGCReKick(ctx, tenantID, "org-x", WorkFileGC, WorkFileGC, 0)
	m.logFileGCReKick(ctx, tenantID, "org-x", WorkFileGC, WorkFileGC, 0)
	m.logFileGCReKick(ctx, tenantID, "org-x", WorkSemantic, 0, 0) // no FileGC bit

	entries := recorded.FilterMessage("tenant_worker_rekick").All()
	if len(entries) != 1 {
		t.Fatalf("rekick log entries = %d, want 1", len(entries))
	}
	fields := entries[0].ContextMap()
	if fields["tenant_id"] != tenantID {
		t.Fatalf("tenant_id = %v, want %s", fields["tenant_id"], tenantID)
	}
	if entries[0].Level != zap.InfoLevel {
		t.Fatalf("rekick log level = %v, want info", entries[0].Level)
	}
}