package server

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/pingcap/failpoint"
	"go.uber.org/zap"

	"github.com/mem9-ai/drive9/pkg/backend"
	"github.com/mem9-ai/drive9/pkg/datastore"
	"github.com/mem9-ai/drive9/pkg/embedding"
	"github.com/mem9-ai/drive9/pkg/journal"
	"github.com/mem9-ai/drive9/pkg/logger"
	"github.com/mem9-ai/drive9/pkg/meta"
	"github.com/mem9-ai/drive9/pkg/metrics"
	"github.com/mem9-ai/drive9/pkg/semantic"
	"github.com/mem9-ai/drive9/pkg/tenant"
)

const (
	defaultTenantWorkers              = 1
	defaultTenantPollInterval         = 200 * time.Millisecond
	defaultTenantLeaseDuration        = 30 * time.Second
	defaultTenantRetryBaseDelay       = 200 * time.Millisecond
	defaultTenantRetryMaxDelay        = 30 * time.Second
	defaultTenantPerTenantConcurrency = 1
	tenantLocalID                     = "local"
	// tenantKickQueueCapacity bounds buffered kicks; overflow kicks are left
	// pending in kickPending and re-enqueued by flushDelayedKicks on the next
	// workerLoop ticker tick. The safety-net scan remains the durable fallback
	// for claiming queued tasks.
	tenantKickQueueCapacity = 256
	// tenantKickDrainLimit caps tasks drained per kick so one busy tenant
	// cannot monopolize a worker; the tenant is re-kicked to continue after
	// other pending kicks get their turn.
	tenantKickDrainLimit = 8
	// fileGCRecoverLimit bounds expired file_gc lease recovery per kick.
	fileGCRecoverLimit = 100
	// fileGCDrainBatchSize caps file_gc tasks drained per kick.
	fileGCDrainBatchSize = 100
	// fileGCMismatchReKickMaxAttempts bounds delayed FileGC re-kicks after
	// claim-empty-with-backlog so a persistent SKIP LOCKED / COUNT mismatch
	// cannot tight-loop.
	fileGCMismatchReKickMaxAttempts = 4
	// semanticRecoverLimit bounds expired semantic task lease recovery per kick.
	semanticRecoverLimit = 64
	// defaultTenantMaintenanceInterval is the throttle interval for piggybacked
	// maintenance (fs_events cleanup + file-count observation) per tenant.
	// Semantic queue observation is NOT on this cadence; see
	// tenantSemanticObserveInterval in tenant_worker_observe.go.
	defaultTenantMaintenanceInterval = 5 * time.Minute
	// tenantWorkerSlotBusyLogInterval rate-limits per-tenant slot_busy warnings
	// so a high-churn write path cannot flood logs when the slot is wedged.
	tenantWorkerSlotBusyLogInterval = 30 * time.Second
	// tenantWorkerSlotLongHold is the hold duration above which a kick logs a
	// warning. Used to surface delete-reconcile / acquire hangs that starve
	// FileGC for one tenant.
	tenantWorkerSlotLongHold = 30 * time.Second
)

// DefaultDeleteReconcileInterval is the default throttle interval for the
// per-tenant delete-orphan reconciliation round (design doc
// recursive-delete-batched-design.md §4.2.1).
const DefaultDeleteReconcileInterval = 10 * time.Minute

// tenantDeleteReconcileInterval is a package-level variable so tests can
// shrink it and main can override it from the environment via
// SetDeleteReconcileInterval.
var tenantDeleteReconcileInterval = DefaultDeleteReconcileInterval

// SetDeleteReconcileInterval overrides the per-tenant delete-orphan
// reconciliation throttle interval for this process. Non-positive values
// keep the current interval. Startup-only: it writes a package-level
// variable without synchronization, so call it once before NewWithConfig
// (wired from DRIVE9_DELETE_RECONCILE_INTERVAL in cmd/drive9-server),
// never concurrently with a running worker.
func SetDeleteReconcileInterval(d time.Duration) {
	if d > 0 {
		tenantDeleteReconcileInterval = d
	}
}

// fileGCMismatchReKickBase is the first delayed FileGC re-kick after
// claim-empty-with-backlog. Tests may shrink it. Backoff doubles each
// attempt up to fileGCMismatchReKickMax.
var fileGCMismatchReKickBase = 5 * time.Second
var fileGCMismatchReKickMax = 30 * time.Second

// drainFileGCClaimableAfterEmptyTestHook, when non-nil, replaces
// ObserveClaimableFileGCTasks after an empty claim (tests only).
var drainFileGCClaimableAfterEmptyTestHook func() (n int64, oldest time.Time, err error)

// deleteReconcilePersistLeadTime is how far in the future RemoveAllCtx's
// fail-closed pre-delete marker is due. The marker is a crash-recovery record;
// the post-delete notifier refresh never pulls a future-dated marker earlier
// (it must not strip another in-flight delete's protection), and the lead
// prevents a poller scan from kicking a sweep off the pre-marker while the
// delete is still in flight (a sweep's cutoff clear must never remove a marker
// that predates delete completion). The default is kept strictly larger than
// the recursive-delete wall-clock budget plus poller/scan and clock-skew
// margins, so a delete cannot outlive its own crash-recovery protection.
// A package-level variable so tests can shrink it.
var deleteReconcilePersistLeadTime = 2*datastore.DeleteDirMaxDuration() +
	2*defaultTenantOutboxCursorFlushInterval + time.Minute

var tenantWorkerAllowedEmbedTaskTypes = []semantic.TaskType{semantic.TaskTypeEmbed, semantic.TaskTypeEmbedBackfill}

func appManagedTenantTaskTypes(embedder embedding.Client) []semantic.TaskType {
	if embedder == nil {
		return nil
	}
	return tenantWorkerAllowedEmbedTaskTypes
}

// TenantWorkerOptions controls unified background task processing.
type TenantWorkerOptions struct {
	// Workers is the number of worker goroutines consuming kicks.
	Workers int
	// PollInterval is the idle wait for the single-tenant fallback ticker.
	PollInterval time.Duration
	// LeaseDuration is the base semantic task lease window used by claim/renew.
	LeaseDuration time.Duration
	// RetryBaseDelay is the initial backoff for semantic retry scheduling.
	RetryBaseDelay time.Duration
	// RetryMaxDelay is the cap for exponential retry backoff.
	RetryMaxDelay time.Duration
	// PerTenantConcurrency limits concurrent tasks per tenant.
	PerTenantConcurrency int
	// MaintenanceInterval throttles piggybacked maintenance per tenant.
	MaintenanceInterval time.Duration
	// FSEventsRetention is how long fs_events rows are kept before the
	// piggybacked maintenance sweep prunes them (backstop for the lazy
	// write-path sweep). Defaults to defaultFSEventsRetention.
	FSEventsRetention time.Duration
	// SweepSharedFSEvents runs the shared-pool fs_events retention sweep for
	// the pool the given store belongs to. Wired from NewWithConfig to the
	// server-level helper (detached; throttled cluster-wide), so the
	// piggyback path and the write path share one throttle. Nil disables the
	// pool sweep (the piggyback branch skips shared stores).
	SweepSharedFSEvents func(store *datastore.Store, tidbCloudOrgID string)
}

func (o *TenantWorkerOptions) normalize() {
	if o.Workers <= 0 {
		o.Workers = defaultTenantWorkers
	}
	if o.PollInterval <= 0 {
		o.PollInterval = defaultTenantPollInterval
	}
	if o.LeaseDuration <= 0 {
		o.LeaseDuration = defaultTenantLeaseDuration
	}
	if o.RetryBaseDelay <= 0 {
		o.RetryBaseDelay = defaultTenantRetryBaseDelay
	}
	if o.RetryMaxDelay <= 0 {
		o.RetryMaxDelay = defaultTenantRetryMaxDelay
	}
	if o.RetryMaxDelay < o.RetryBaseDelay {
		o.RetryMaxDelay = o.RetryBaseDelay
	}
	if o.PerTenantConcurrency <= 0 {
		o.PerTenantConcurrency = defaultTenantPerTenantConcurrency
	}
	if o.MaintenanceInterval <= 0 {
		o.MaintenanceInterval = defaultTenantMaintenanceInterval
	}
	if o.FSEventsRetention <= 0 {
		o.FSEventsRetention = defaultFSEventsRetention
	}
}

// TenantWorkerWillRun reports whether NewWithConfig would construct a non-nil
// tenant worker manager for cfg.
func TenantWorkerWillRun(cfg Config) bool {
	return newTenantWorkerManager(cfg.Backend, cfg.Meta, cfg.Pool, cfg.SemanticEmbedder, cfg.TenantWorkers, cfg.TenantMaintenanceInterval) != nil
}

// ValidateDurableAsyncExtractRequiresTenantWorker returns an error when async
// image or audio extraction runtimes are enabled on the backend template but
// the tenant worker would not start for cfg.
func ValidateDurableAsyncExtractRequiresTenantWorker(cfg Config, template backend.Options, localTiDBAutoOnly bool) error {
	willWire := backend.AsyncImageExtractWillWireRuntime(template.AsyncImageExtract) ||
		backend.AsyncAudioExtractWillWireRuntime(template.AsyncAudioExtract)
	if !willWire {
		return nil
	}
	if localTiDBAutoOnly && !template.DatabaseAutoEmbedding {
		return nil
	}
	if TenantWorkerWillRun(cfg) {
		return nil
	}
	return fmt.Errorf("tenant worker would not start but durable async image/audio extract is enabled; configure DRIVE9_EMBED_* for app-managed embedding or fix worker/task-type routing so img_extract_text and audio_extract_text can be claimed")
}

// kickMsg carries a tenant ID and accumulated work mask to a worker goroutine.
type kickMsg struct {
	tenantID       string
	tidbCloudOrgID string
	workMask       int
}

type pendingKick struct {
	tidbCloudOrgID string
	workMask       int
}

// tenantWorkerManager is the unified worker that processes kicks from the
// tenant_notify_outbox poller. Each kick carries a work_mask selecting which
// work types to drain: semantic tasks, file_gc tasks, quota outbox rows. The
// manager also recovers expired leases and runs piggyback maintenance
// (fs_events cleanup + observation metrics) with a per-tenant throttle.
//
// In single-tenant (fallback) mode, a workerLoop ticker polls the fallback
// backend for all work types on PollInterval.
type tenantWorkerManager struct {
	fallback *backend.Dat9Backend
	meta     *meta.Store
	pool     *tenant.Pool
	embedder embedding.Client
	opts     TenantWorkerOptions

	// extractResolver resolves the effective per-tenant image extract runtime
	// at task execution time. Nil means no resolver is wired (single-tenant
	// local server): img_extract_text tasks fall back to the process-default
	// backend runtime.
	extractResolver *extractConfigResolver
	// imageExtractGlobalDisabled is the cluster image-extract kill switch,
	// shared with the Server. While on, workers stop claiming NEW
	// img_extract_text tasks (claim-time filter) and an already-claimed task
	// that has not called the provider is deferred with global_disabled. File
	// writes keep enqueueing tasks; no task is acked because of the switch.
	// Nil means image extract is allowed.
	imageExtractGlobalDisabled *atomic.Bool

	// ownsTenant reports shard ownership for FileGC gauge SET/CLEAR. Nil
	// means this process owns every tenant (single-pod / tests).
	ownsTenant func(string) bool

	mu          sync.Mutex
	inflight    map[string]int
	processing  int
	kickPending map[string]pendingKick // tenantID → accumulated work_mask + best-known org id

	kicks chan kickMsg

	lastMaintenance        map[string]time.Time
	lastSemanticObserve    map[string]time.Time   // tenantID -> last semantic queue observation (own throttle, not MaintenanceInterval)
	lastReconcile          map[string]time.Time   // tenantID -> last delete-orphan reconciliation round
	reconcileTimers        map[string]*time.Timer // tenantID -> one delayed WorkDeleteReconcile re-kick
	fileGCMismatchTimers   map[string]*time.Timer // tenantID -> delayed FileGC re-kick after claim/observe mismatch
	fileGCMismatchAttempts map[string]int         // tenantID -> mismatch re-kick attempts in the current streak
	storageTierTimers      map[string]*time.Timer // tenantID -> one delayed WorkStorageTierScan re-kick
	storageTierOwnerID     string                 // process-unique component of persisted scan lease owners
	storageTierScanRun     uint64                 // monotonically identifies one local scan lease owner
	reconcileState         map[string]*datastore.DeleteReconcileState
	semanticMetricOrg      map[string]string // tenantID -> org used by the last semantic gauge observation
	// lastRateLimitedLog rate-limits Warn logs keyed by tenantID+"\x00"+name
	// (slot_busy, claim_empty_with_backlog, slot_held_long).
	lastRateLimitedLog map[string]time.Time

	cancel  context.CancelFunc
	wg      sync.WaitGroup
	stopped bool // protected by mu; blocks delayed callbacks after Stop
}

type tenantTarget struct {
	tenantID         string
	tidbCloudOrgID   string
	backend          *backend.Dat9Backend
	store            *datastore.Store
	allowedTaskTypes []semantic.TaskType
	release          func()
}

func (t *tenantTarget) metricOrgID() string {
	if t == nil {
		return defaultTenantMetricTiDBCloudOrgID
	}
	return normalizeTenantMetricTiDBCloudOrgID(t.tidbCloudOrgID)
}

func newTenantWorkerManager(fallback *backend.Dat9Backend, metaStore *meta.Store, pool *tenant.Pool, embedder embedding.Client, opts TenantWorkerOptions, maintenanceInterval time.Duration) *tenantWorkerManager {
	hasMultiTenant := metaStore != nil && pool != nil
	if fallback == nil && !hasMultiTenant {
		return nil
	}
	m := &tenantWorkerManager{
		fallback: fallback,
		meta:     metaStore,
		pool:     pool,
		embedder: embedder,
	}
	if maintenanceInterval > 0 {
		opts.MaintenanceInterval = maintenanceInterval
	}
	// The unified tenant worker handles semantic, file_gc, and quota work.
	// It is viable whenever there is a multi-tenant pool or a fallback backend.
	viable := hasMultiTenant || fallback != nil
	if !viable {
		return nil
	}
	// In single-tenant (fallback-only) mode, semantic work requires an
	// embedder or async-extract support; file_gc and quota remain viable
	// regardless, so we do not return nil for file_gc/quota-only deployments.
	opts.normalize()
	m.opts = opts
	m.inflight = make(map[string]int)
	m.kickPending = make(map[string]pendingKick)
	m.lastMaintenance = make(map[string]time.Time)
	m.lastSemanticObserve = make(map[string]time.Time)
	m.lastReconcile = make(map[string]time.Time)
	m.reconcileState = make(map[string]*datastore.DeleteReconcileState)
	m.reconcileTimers = make(map[string]*time.Timer)
	m.fileGCMismatchTimers = make(map[string]*time.Timer)
	m.fileGCMismatchAttempts = make(map[string]int)
	m.storageTierTimers = make(map[string]*time.Timer)
	m.storageTierOwnerID = journal.NewID("tier-scan")
	m.semanticMetricOrg = make(map[string]string)
	m.lastRateLimitedLog = make(map[string]time.Time)
	m.kicks = make(chan kickMsg, tenantKickQueueCapacity)
	return m
}

// Kick prompts a worker to process work for tenantID. The workMask is
// OR-accumulated with any pending kick for the same tenant so a burst of
// different work types coalesces into one kick. Best-effort: duplicates
// collapse while one is pending, and the kick is left pending (delayed) when
// the buffer is full — flushDelayedKicks re-enqueues it on the next ticker
// tick, and the safety-net scan remains the durable fallback path.
func (m *tenantWorkerManager) Kick(tenantID string, workMask int) {
	m.KickWithOrg(tenantID, "", workMask)
}

// KickWithOrg is Kick with an already-resolved TiDB Cloud org label. Callers
// without a resolved org should pass an empty string and let metrics normalize
// it to the shared guest label.
func (m *tenantWorkerManager) KickWithOrg(tenantID, tidbCloudOrgID string, workMask int) {
	if m == nil || tenantID == "" || workMask == 0 {
		return
	}
	m.mu.Lock()
	if m.stopped {
		m.mu.Unlock()
		return
	}
	if pending, ok := m.kickPending[tenantID]; ok {
		pending.workMask |= workMask
		pending.tidbCloudOrgID = mergeKickOrgID(pending.tidbCloudOrgID, tidbCloudOrgID)
		m.kickPending[tenantID] = pending
		m.mu.Unlock()
		// Aggregate only (no tenant_id on success/neutral series — #737/#802).
		metrics.RecordTenantOperationWithOrg(tenantID, pending.tidbCloudOrgID, "tenant_worker", "kick", "coalesced", 0)
		// Tenant-attributed FileGC scheduling evidence (default-visible, rate-
		// limited): proves tenant X's WorkFileGC coalesced into a pending kick
		// before processKicked/acquire. Metric above stays tenant-free.
		if pending.workMask&WorkFileGC != 0 &&
			m.shouldRateLimitedLog(tenantID, "kick_coalesced", tenantWorkerSlotBusyLogInterval) {
			logger.Info(context.Background(), "tenant_worker_kick_coalesced",
				zap.String("tenant_id", tenantID),
				zap.Int("work_mask", workMask),
				zap.Int("pending_mask", pending.workMask),
				zap.String("tidbcloud_org_id", pending.tidbCloudOrgID))
		}
		return
	}
	tidbCloudOrgID = strings.TrimSpace(tidbCloudOrgID)
	m.kickPending[tenantID] = pendingKick{tidbCloudOrgID: tidbCloudOrgID, workMask: workMask}
	m.mu.Unlock()
	select {
	case m.kicks <- kickMsg{tenantID: tenantID, tidbCloudOrgID: tidbCloudOrgID, workMask: workMask}:
		metrics.RecordTenantOperationWithOrg(tenantID, tidbCloudOrgID, "tenant_worker", "kick", "queued", 0)
	default:
		// Channel full: leave kickPending so flushDelayedKicks (ticked from
		// workerLoop) re-enqueues when the channel has space. The work mask is
		// already recorded in kickPending above; do NOT clear it.
		metrics.RecordTenantOperationWithOrg(tenantID, tidbCloudOrgID, "tenant_worker", "kick", "delayed", 0)
		// Channel-full backpressure is rare and tenant-visible at Warn so
		// production (default info+) can prove WorkFileGC was delayed before
		// processKicked without adding tenant labels to success metrics.
		if workMask&WorkFileGC != 0 &&
			m.shouldRateLimitedLog(tenantID, "kick_delayed", tenantWorkerSlotBusyLogInterval) {
			logger.Warn(context.Background(), "tenant_worker_kick_delayed",
				zap.String("tenant_id", tenantID),
				zap.Int("work_mask", workMask),
				zap.String("tidbcloud_org_id", tidbCloudOrgID))
		}
	}
}

// ForgetTenant drops process-local scheduler bookkeeping after the tenant
// lifecycle has durably entered deletion. A queued kick may still be present
// in the channel, but processKicked will reject it against the persisted
// non-active status and clear it again.
func (m *tenantWorkerManager) ForgetTenant(tenantID string) {
	if m == nil || tenantID == "" {
		return
	}
	m.mu.Lock()
	delete(m.kickPending, tenantID)
	delete(m.lastMaintenance, tenantID)
	delete(m.lastSemanticObserve, tenantID)
	delete(m.lastReconcile, tenantID)
	delete(m.reconcileState, tenantID)
	if t, ok := m.reconcileTimers[tenantID]; ok {
		t.Stop()
		delete(m.reconcileTimers, tenantID)
	}
	if t, ok := m.fileGCMismatchTimers[tenantID]; ok {
		t.Stop()
		delete(m.fileGCMismatchTimers, tenantID)
	}
	delete(m.fileGCMismatchAttempts, tenantID)
	if t, ok := m.storageTierTimers[tenantID]; ok {
		t.Stop()
		delete(m.storageTierTimers, tenantID)
	}
	delete(m.semanticMetricOrg, tenantID)
	for k := range m.lastRateLimitedLog {
		if strings.HasPrefix(k, tenantID+"\x00") {
			delete(m.lastRateLimitedLog, k)
		}
	}
	m.mu.Unlock()
	metrics.ClearFileGCClaimableBacklog(tenantID)
	// Drop the durable pending marker so a deleted tenant is not re-kicked by
	// the outbox poller's pending scan (kickRef would reject it anyway, but
	// the row would otherwise accumulate).
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	m.clearDeleteReconcilePending(ctx, tenantID)
	cancel()
}

func mergeKickOrgID(current, next string) string {
	next = strings.TrimSpace(next)
	if next != "" {
		return next
	}
	return current
}

func (m *tenantWorkerManager) clearKickPending(tenantID string) {
	m.mu.Lock()
	delete(m.kickPending, tenantID)
	m.mu.Unlock()
}

// takePendingKick returns the accumulated work mask and best-known org id for
// a tenant, and clears the pending entry. Called by the worker before
// dispatching so a kick arriving during processing triggers a fresh kick.
func (m *tenantWorkerManager) takePendingKick(tenantID string) pendingKick {
	m.mu.Lock()
	pending := m.kickPending[tenantID]
	delete(m.kickPending, tenantID)
	m.mu.Unlock()
	return pending
}

// peekPendingWorkMask returns the pending work mask without clearing it.
func (m *tenantWorkerManager) peekPendingWorkMask(tenantID string) int {
	m.mu.Lock()
	mask := m.kickPending[tenantID].workMask
	m.mu.Unlock()
	return mask
}

func (m *tenantWorkerManager) takePendingWorkMask(tenantID string) int {
	pending := m.takePendingKick(tenantID)
	return pending.workMask
}

// flushDelayedKicks re-enqueues kicks that were left pending when the kicks
// channel was full. Called on each workerLoop ticker tick so delayed work is
// delivered once the channel drains. Entries still pending after a successful
// send are consumed by the receiving worker via takePendingWorkMask.
func (m *tenantWorkerManager) flushDelayedKicks() {
	// Snapshot pending entries under the lock, then attempt non-blocking sends
	// outside the lock to avoid holding it during channel ops.
	m.mu.Lock()
	if len(m.kickPending) == 0 {
		m.mu.Unlock()
		return
	}
	pending := make(map[string]pendingKick, len(m.kickPending))
	for tenantID, entry := range m.kickPending {
		pending[tenantID] = entry
	}
	m.mu.Unlock()
	for tenantID, entry := range pending {
		select {
		case m.kicks <- kickMsg{tenantID: tenantID, tidbCloudOrgID: entry.tidbCloudOrgID, workMask: entry.workMask}:
			// Successfully (re-)queued. Leave the kickPending entry in place:
			// the worker merges it via takePendingWorkMask after claiming the
			// slot, so any further coalesced kicks are not lost.
			metrics.RecordTenantOperationWithOrg(tenantID, entry.tidbCloudOrgID, "tenant_worker", "kick", "flushed", 0)
		default:
			// Still full; leave pending for the next tick.
		}
	}
}

func (m *tenantWorkerManager) Start(ctx context.Context) {
	if m == nil || m.cancel != nil {
		return
	}
	workerCtx, cancel := context.WithCancel(backgroundWithTrace(ctx))
	m.cancel = cancel
	m.mu.Lock()
	m.stopped = false
	m.mu.Unlock()
	metrics.SetModuleAvailability("semantic_worker", true)
	metrics.RecordGauge("semantic_worker", "workers", float64(m.opts.Workers))
	metrics.RecordGauge("semantic_worker", "inflight", 0)
	for i := 0; i < m.opts.Workers; i++ {
		m.wg.Add(1)
		go m.workerLoop(workerCtx, i+1)
	}
	logger.Info(workerCtx, "tenant_worker_manager_started",
		zap.Int("workers", m.opts.Workers),
		zap.Duration("poll_interval", m.opts.PollInterval),
		zap.Duration("lease_duration", m.opts.LeaseDuration),
		zap.Duration("maintenance_interval", m.opts.MaintenanceInterval))
}

func (m *tenantWorkerManager) Stop() {
	if m == nil || m.cancel == nil {
		return
	}
	m.mu.Lock()
	m.stopped = true
	for tenantID, t := range m.reconcileTimers {
		t.Stop()
		delete(m.reconcileTimers, tenantID)
	}
	for tenantID, t := range m.fileGCMismatchTimers {
		t.Stop()
		delete(m.fileGCMismatchTimers, tenantID)
	}
	m.fileGCMismatchAttempts = make(map[string]int)
	for tenantID, t := range m.storageTierTimers {
		t.Stop()
		delete(m.storageTierTimers, tenantID)
	}
	m.mu.Unlock()
	m.cancel()
	m.wg.Wait()
	m.cancel = nil
	m.mu.Lock()
	defer m.mu.Unlock()
	m.processing = 0
	metrics.SetModuleAvailability("semantic_worker", false)
	metrics.RecordGauge("semantic_worker", "workers", 0)
	metrics.RecordGauge("semantic_worker", "inflight", 0)
}

func (m *tenantWorkerManager) workerLoop(ctx context.Context, workerID int) {
	defer m.wg.Done()
	// Single-tenant fallback: poll the fallback backend on PollInterval when
	// no kicks arrive (no meta store / no pool). This keeps the local backend
	// drained without depending on the outbox poller.
	ticker := time.NewTicker(m.opts.PollInterval)
	defer ticker.Stop()
	for {
		if ctx.Err() != nil {
			logger.Info(ctx, "tenant_worker_stopped", zap.Int("worker_id", workerID))
			return
		}
		// Recover around each loop body so panics in processKicked *or*
		// pollFallbackOnce cannot permanently kill the only worker (Workers=1).
		// If a panic occurs, wait one PollInterval before the next iteration so
		// a deterministic fallback panic cannot spin a tight CPU/log loop
		// (shouldPollFallback stays true and would otherwise re-enter immediately).
		panicked := false
		func() {
			defer func() {
				if rec := recover(); rec != nil {
					panicked = true
					logger.Error(ctx, "tenant_worker_loop_panic",
						zap.Int("worker_id", workerID),
						zap.Any("panic", rec),
						zap.Stack("stack"))
					metrics.RecordOperation("tenant_worker", "worker_loop", "panic", 0)
				}
			}()
			// In single-tenant mode with a fallback backend, drain it proactively.
			if m.shouldPollFallback() {
				processed := m.pollFallbackOnce(ctx)
				if processed {
					return
				}
			}
			select {
			case <-ctx.Done():
				return
			case msg := <-m.kicks:
				m.processKicked(ctx, msg.tenantID, msg.tidbCloudOrgID, msg.workMask)
			case <-ticker.C:
				// Re-enqueue delayed kicks now that the channel may have space.
				m.flushDelayedKicks()
			}
		}()
		if panicked {
			// Fresh timer per panic: the long-lived ticker may already hold a
			// tick when the panicking work ran longer than PollInterval, which
			// would make a ticker wait return immediately (no real backoff).
			// Measure the backoff from recovery instead.
			backoff := time.NewTimer(m.opts.PollInterval)
			select {
			case <-ctx.Done():
				if !backoff.Stop() {
					select {
					case <-backoff.C:
					default:
					}
				}
				return
			case <-backoff.C:
			}
		}
	}
}

// pollFallbackOnce drains the local fallback backend for one round of semantic,
// file_gc, and quota work. Returns true if any task was processed (caller should
// loop immediately), false if all queues are empty.
func (m *tenantWorkerManager) pollFallbackOnce(ctx context.Context) bool {
	if pollFallbackOnceTestHook != nil {
		pollFallbackOnceTestHook(m)
	}
	if m.fallback == nil {
		return false
	}
	if ctx.Err() != nil {
		return false
	}
	target := &tenantTarget{
		tenantID:         tenantLocalID,
		backend:          m.fallback,
		store:            m.fallback.Store(),
		allowedTaskTypes: m.taskTypesForTarget(m.fallback),
		release:          func() {},
	}
	processed := false
	if len(target.allowedTaskTypes) > 0 {
		claimed, _ := m.claimAndProcessOne(ctx, target)
		if claimed {
			processed = true
		}
		// Single-tenant mode has no kick-driven observation; sample on the
		// throttled cadence regardless of the claim outcome so a local tenant
		// both publishes a growing backlog and clears its lag gauge after a
		// drain.
		m.observeTenantThrottled(ctx, target)
	}
	// Drain one file_gc task. ProcessOneFileGCTask returns (false, nil) when
	// the queue is empty — only set processed=true when it actually processed.
	if did, err := m.fallback.ProcessOneFileGCTask(ctx); err == nil && did {
		processed = true
	}
	return processed
}

func (m *tenantWorkerManager) shouldPollFallback() bool {
	if m.fallback == nil {
		return false
	}
	// Only poll fallback proactively in single-tenant mode (no multi-tenant
	// pool). In multi-tenant mode the fallback is not scheduled by kicks.
	return m.meta == nil || m.pool == nil
}

// processKicked resolves a kicked tenant, acquires its backend, and drains all
// selected work types. workMask selects which drains run. After draining, it
// recovers expired leases and runs piggyback maintenance (throttled). The
// tenant is re-kicked if more work remains.
//
// Observability notes (cardinality): aggregate counters use component/operation/
// result without tenant_id. tenant_id appears only on structured logs (and the
// existing failure-only tenant counter path). Slot-busy is rate-limited per
// tenant so churny write paths cannot flood logs.
func (m *tenantWorkerManager) processKicked(ctx context.Context, tenantID, tidbCloudOrgID string, workMask int) {
	// Recover so a panic in one kick cannot permanently kill the worker loop
	// (default Workers=1). Slot release is tracked explicitly for the window
	// before target.release is registered.
	var (
		slotTenant string
		slotHeld   bool
	)
	defer func() {
		if rec := recover(); rec != nil {
			logger.Error(ctx, "tenant_worker_process_kicked_panic",
				zap.String("tenant_id", tenantID),
				zap.Int("work_mask", workMask),
				zap.Any("panic", rec),
				zap.Stack("stack"))
			// "panic" is classified as a failure result for alerting; tenant_id
			// is known here so the failure series can attribute the tenant.
			metrics.RecordTenantOperationCountWithOrg(tenantID, tidbCloudOrgID, "tenant_worker", "process_kicked", "panic")
			if slotHeld {
				m.releaseTenantSlot(slotTenant)
			}
		}
	}()

	if workMask&WorkStorageTierScan != 0 {
		pending := m.takePendingKick(tenantID)
		workMask |= pending.workMask
		tidbCloudOrgID = mergeKickOrgID(tidbCloudOrgID, pending.tidbCloudOrgID)
		m.processWarmStorageTierScan(ctx, tenantID, tidbCloudOrgID)
		if processKickedAfterStorageTierScanTestHook != nil {
			processKickedAfterStorageTierScanTestHook(tenantID)
		}
		workMask &^= WorkStorageTierScan
		if workMask == 0 {
			return
		}
	}

	// Do NOT clear kickPending here — a kick coalesced while this message was
	// in the channel must be consumed after acquiring the tenant slot.
	ref, ok, refErr := m.kickRef(ctx, tenantID, workMask)
	if !ok {
		transient := refErr != nil && !errors.Is(refErr, meta.ErrNotFound)
		// Definitive rejects clear pending work. Transient lookup errors are
		// re-seeded below when the scan prologue consumed a coalesced mask.
		if transient {
			logger.Warn(ctx, "tenant_worker_kick_ref_failed",
				zap.String("tenant_id", tenantID),
				zap.Int("work_mask", workMask),
				zap.Bool("cleared_pending", false),
				zap.Error(refErr))
			metrics.RecordTenantOperationCountWithOrg(tenantID, tidbCloudOrgID, "tenant_worker", "kick_ref", "error")
		} else {
			metrics.RecordTenantOperationCountWithOrg(tenantID, tidbCloudOrgID, "tenant_worker", "kick_ref", "rejected")
		}
		if transient {
			// Include work that coalesced while the storage scan ran. Taking it
			// before re-seeding also preserves channel-full delayed kicks.
			pending := m.takePendingKick(tenantID)
			workMask |= pending.workMask
			tidbCloudOrgID = mergeKickOrgID(tidbCloudOrgID, pending.tidbCloudOrgID)
		} else {
			m.clearKickPending(tenantID)
		}
		if transient && workMask != 0 {
			// The scan prologue may have consumed the coalesced mask. A
			// transient ref lookup failure must preserve it for retry.
			m.KickWithOrg(tenantID, tidbCloudOrgID, workMask)
		}
		if workMask&WorkDeleteReconcile != 0 && (refErr == nil || errors.Is(refErr, meta.ErrNotFound)) {
			// The tenant is definitively not schedulable (missing, suspended,
			// deleted, or has no work types): drop its durable pending marker
			// so the poller scan stops re-kicking it (same lifecycle intent
			// as ForgetTenant). Transient lookup errors must NOT clear the
			// marker — the poller scan retries those.
			m.clearDeleteReconcilePending(ctx, tenantID)
		}
		return
	}
	if !m.tryClaimTenantSlot(ref.id) {
		// Another worker holds the slot. The storage-scan pre-dispatch above may
		// already have consumed kickPending, so restore all non-scan work before
		// returning. Otherwise a coalesced FileGC/semantic kick can be lost.
		if workMask != 0 {
			m.KickWithOrg(ref.id, tidbCloudOrgID, workMask)
		}
		inflight := m.tenantSlotInflight(ref.id)
		if m.shouldRateLimitedLog(ref.id, "slot_busy", tenantWorkerSlotBusyLogInterval) {
			logger.Warn(ctx, "tenant_worker_slot_busy",
				zap.String("tenant_id", ref.id),
				zap.Int("work_mask", workMask),
				zap.Int("inflight", inflight),
				zap.String("tidbcloud_org_id", tidbCloudOrgID))
		}
		// Aggregate only: result "busy" is intentionally not a tenant-failure
		// series (keeps cardinality policy from #737/#802).
		metrics.RecordTenantOperationCountWithOrg(ref.id, tidbCloudOrgID, "tenant_worker", "slot", "busy")
		// FileGC-specific: attribute sustained slot starvation on the tenant
		// failure series so alerts can page before quota stays inflated.
		// Include pending coalesced mask: FileGC may sit only in kickPending
		// while this kick carried another work type.
		if workMask&WorkFileGC != 0 || m.peekPendingWorkMask(ref.id)&WorkFileGC != 0 {
			metrics.RecordTenantOperationCountWithOrg(ref.id, tidbCloudOrgID, "file_gc", "sched", "slot_busy_error")
		}
		return
	}
	slotTenant = ref.id
	slotHeld = true
	slotStart := time.Now()
	// Test-only hook (nil in production): runs while slotHeld=true so panic
	// recovery can be exercised end-to-end without failpoint instrumentation.
	if processKickedAfterSlotClaimTestHook != nil {
		processKickedAfterSlotClaimTestHook(ref.id)
	}

	// Merge any work that coalesced while this kick was queued.
	pending := m.takePendingKick(tenantID)
	workMask |= pending.workMask
	tidbCloudOrgID = mergeKickOrgID(tidbCloudOrgID, pending.tidbCloudOrgID)
	target, err := m.targetForRef(ctx, ref, tidbCloudOrgID)
	if err != nil {
		// Mask was already taken from kickPending; restore it so a transient
		// tenant open failure cannot drop coalesced work.
		// Suffix _error so drive9_tenant_operation_failures_total attributes the tenant.
		logger.Warn(ctx, "tenant_worker_kick_open_store_failed",
			zap.String("tenant_id", ref.id),
			zap.Int("retry_work_mask", workMask),
			zap.Error(err))
		metrics.RecordTenantOperationCountWithOrg(ref.id, tidbCloudOrgID, "tenant_worker", "kick", "retry_acquire_error")
		if workMask != 0 {
			m.KickWithOrg(ref.id, tidbCloudOrgID, workMask)
		}
		m.releaseTenantSlot(ref.id)
		slotHeld = false
		return
	}
	orgID := target.metricOrgID()
	poolRelease := target.release
	// Release slot first so a panic in metrics/logging cannot wedge concurrency=1.
	// Then release the pool backend.
	target.release = func() {
		m.releaseTenantSlot(ref.id)
		held := time.Since(slotStart)
		// Duration histogram is aggregate (no tenant_id on success series).
		metrics.RecordTenantOperationWithOrg(ref.id, orgID, "tenant_worker", "slot_hold", "ok", held)
		if held >= tenantWorkerSlotLongHold &&
			m.shouldRateLimitedLog(ref.id, "slot_held_long", tenantWorkerSlotBusyLogInterval) {
			logger.Warn(ctx, "tenant_worker_slot_held_long",
				zap.String("tenant_id", ref.id),
				zap.Int("work_mask", workMask),
				zap.Duration("held", held),
				zap.String("tidbcloud_org_id", orgID))
		}
		if poolRelease != nil {
			poolRelease()
		}
	}
	// Slot release is now owned by target.release (including panic unwind).
	slotHeld = false
	defer target.release()

	// FileGC gauges follow Claim, not COUNT. An already-queued kick can
	// outlive a ring change, so re-check ownership before drain: a SET
	// here would recreate leftover series the ownership-loss callback
	// just deleted. Do not ObserveClaimable (COUNT) before Claim — that
	// republishes phantom depth during semantic work and on mismatch re-kicks.
	if workMask&WorkFileGC != 0 && !m.ownsFileGC(ref.id) {
		workMask &^= WorkFileGC
		metrics.ClearFileGCClaimableBacklog(ref.id)
		m.clearFileGCMismatchBackoff(ref.id)
	}

	// Drain semantic tasks (if selected and allowed).
	var reKickMask int
	semanticMask := workMask & WorkSemantic
	if semanticMask != 0 && len(target.allowedTaskTypes) > 0 {
		semanticDrained := 0
		queueEmpty := false
		for range tenantKickDrainLimit {
			if ctx.Err() != nil {
				return
			}
			claimed, empty := m.claimAndProcessOne(ctx, target)
			if !claimed {
				queueEmpty = empty
				break
			}
			semanticDrained++
		}
		if queueEmpty {
			// The claimable queue just drained to empty: refresh the tenant's
			// semantic observation immediately so a previously exported
			// queue_lag_seconds / dead_lettered gauge clears now instead of
			// waiting out the observation throttle — a recovered tenant must
			// resolve its lag alert promptly.
			m.observeTenantNow(ctx, target)
		}
		if semanticDrained >= tenantKickDrainLimit {
			reKickMask |= WorkSemantic
		}
	}

	// Drain file_gc tasks (if selected).
	if workMask&WorkFileGC != 0 {
		if m.drainFileGC(ctx, target) {
			reKickMask |= WorkFileGC
		}
	}
	// WorkDeleteReconcile needs no dedicated drain: piggybackMaintenance below
	// runs the delete-orphan reconciliation sweep, but only on the 5-minute
	// maintenance cadence. An explicit delete must not depend on that cadence,
	// so a WorkDeleteReconcile kick requests the sweep directly (honoring the
	// per-tenant reconcile throttle, with a delayed re-kick if throttled).
	if workMask&WorkDeleteReconcile != 0 {
		m.requestDeleteReconcile(ctx, target)
	}

	// Recover expired leases for all work types (cheap, runs on every kick).
	m.recoverExpired(ctx, target)

	// Semantic queue observation has its own per-tenant throttle, independent
	// of MaintenanceInterval: queue_lag_seconds feeds lag alerting and must
	// stay fresh even when maintenance is configured with a long interval.
	m.observeTenantThrottled(ctx, target)

	// Piggyback maintenance: fs_events cleanup, throttled per tenant by
	// MaintenanceInterval.
	m.piggybackMaintenance(ctx, target)

	// Re-kick if there's pending work accumulated during processing, or if a
	// drain hit its cap (more work likely remains for that type).
	pending = m.takePendingKick(tenantID)
	reKick := pending.workMask | reKickMask
	if reKick != 0 {
		// Aggregate counter only (no tenant_id on success series).
		metrics.RecordTenantOperationCountWithOrg(tenantID, orgID, "tenant_worker", "rekick", "ok")
		// Default-visible, rate-limited FileGC re-kick evidence (was Debug and
		// invisible at DRIVE9_LOG_LEVEL=info). Complements pre-processKicked
		// coalesced/delayed logs for the full scheduler path.
		m.logFileGCReKick(ctx, tenantID, orgID, reKick, reKickMask, pending.workMask)
		m.KickWithOrg(tenantID, mergeKickOrgID(orgID, pending.tidbCloudOrgID), reKick)
	}
}

const (
	storageTierScanPageSize       = 500
	storageTierScanReleaseTimeout = 5 * time.Second
)

func storageTierScanReleaseContext() (context.Context, context.CancelFunc) {
	return context.WithTimeout(context.Background(), storageTierScanReleaseTimeout)
}

func (m *tenantWorkerManager) processWarmStorageTierScan(ctx context.Context, tenantID, tidbCloudOrgID string) {
	if m == nil || m.meta == nil || m.pool == nil || tenantID == "" || tenantID == tenantLocalID {
		return
	}
	tenantRecord, err := m.meta.GetTenant(ctx, tenantID)
	if err != nil || tenantRecord.Status != meta.TenantActive {
		return
	}
	b, release, ok := m.pool.AcquireCached(tenantRecord)
	if !ok {
		// The durable job remains pending. A later natural cold-open schedules
		// another scan signal; this path never activates the tenant DB.
		return
	}
	defer release()

	owner := m.nextStorageTierScanLeaseOwner()
	job, claimed, err := m.meta.ClaimStorageTierScanJob(ctx, tenantID, owner, m.opts.LeaseDuration)
	if err != nil || !claimed {
		return
	}
	releaseForRetry := func(scanErr error) {
		releaseCtx, cancel := storageTierScanReleaseContext()
		defer cancel()
		if err := m.meta.ReleaseStorageTierScanJob(releaseCtx, tenantID, owner, scanErr); err != nil {
			logger.Warn(ctx, "storage_tier_scan_release_failed", zap.String("tenant_id", tenantID), zap.Error(err))
			return
		}
		m.rekickStorageTierScan(tenantID, tidbCloudOrgID)
	}
	mutationState, err := m.meta.StorageTierScanWatermarkState(ctx, tenantID, job.Watermark)
	if err != nil {
		releaseForRetry(err)
		return
	}
	if mutationState.Failed > 0 {
		if err := m.meta.BlockStorageTierScanJob(ctx, tenantID, owner, "storage tier mutation failed"); err != nil {
			releaseForRetry(err)
		}
		return
	}
	if mutationState.Pending > 0 {
		releaseForRetry(nil)
		return
	}
	if err := m.meta.InitializeStorageTierCounters(ctx, tenantID, owner); err != nil {
		releaseForRetry(err)
		return
	}
	completed, err := false, error(nil)
	if job.Phase == meta.StorageTierScanPhaseLayer {
		cursor := datastore.FSLayerStorageTierCursor{
			LayerID: job.Cursor, PathHash: job.LayerPathHash, EntrySeq: job.LayerEntrySeq,
		}
		page, nextCursor, scanErr := b.Store().ListFSLayerStorageTiers(ctx, cursor, storageTierScanPageSize)
		if scanErr != nil {
			releaseForRetry(scanErr)
			return
		}
		layers := make([]meta.StorageTierScanLayer, 0, len(page))
		for _, item := range page {
			layers = append(layers, meta.StorageTierScanLayer{
				StorageTierLayerCursor: meta.StorageTierLayerCursor{
					LayerID: item.LayerID, PathHash: item.PathHash, EntrySeq: item.EntrySeq,
				},
				SizeBytes: item.SizeBytes, StorageType: meta.StorageType(item.StorageType),
			})
		}
		completed, err = m.meta.ApplyStorageTierLayerScanPage(ctx, tenantID, owner, meta.StorageTierLayerCursor{
			LayerID: nextCursor.LayerID, PathHash: nextCursor.PathHash, EntrySeq: nextCursor.EntrySeq,
		}, layers)
	} else {
		page, nextCursor, scanErr := b.Store().ListConfirmedFileStorageTiers(ctx, job.Cursor, storageTierScanPageSize)
		if scanErr != nil {
			releaseForRetry(scanErr)
			return
		}
		files := make([]meta.StorageTierScanFile, 0, len(page))
		for _, item := range page {
			files = append(files, meta.StorageTierScanFile{
				FileID: item.FileID, SizeBytes: item.SizeBytes, IsMedia: item.IsMedia,
				StorageType: meta.StorageType(item.StorageType), LayerOwned: item.LayerOwned,
			})
		}
		err = m.meta.ApplyStorageTierScanPage(ctx, tenantID, owner, nextCursor, files)
	}
	if err != nil {
		releaseForRetry(err)
		return
	}
	metrics.RecordTenantOperationWithOrg(tenantID, tidbCloudOrgID, "storage_tier_scan", "page", "ok", 0)
	if !completed {
		m.rekickStorageTierScan(tenantID, tidbCloudOrgID)
	}
}

func (m *tenantWorkerManager) nextStorageTierScanLeaseOwner() string {
	m.mu.Lock()
	if m.storageTierOwnerID == "" {
		m.storageTierOwnerID = journal.NewID("tier-scan")
	}
	m.storageTierScanRun++
	run := m.storageTierScanRun
	ownerID := m.storageTierOwnerID
	m.mu.Unlock()
	return fmt.Sprintf("%s-%d", ownerID, run)
}

func (m *tenantWorkerManager) rekickStorageTierScan(tenantID, tidbCloudOrgID string) {
	m.mu.Lock()
	if m.stopped {
		m.mu.Unlock()
		return
	}
	if m.storageTierTimers == nil {
		m.storageTierTimers = make(map[string]*time.Timer)
	}
	if _, exists := m.storageTierTimers[tenantID]; exists {
		m.mu.Unlock()
		return
	}
	var timer *time.Timer
	timer = time.AfterFunc(time.Second, func() {
		m.mu.Lock()
		if current, ok := m.storageTierTimers[tenantID]; !ok || current != timer {
			m.mu.Unlock()
			return
		}
		delete(m.storageTierTimers, tenantID)
		m.mu.Unlock()
		m.KickWithOrg(tenantID, tidbCloudOrgID, WorkStorageTierScan)
	})
	m.storageTierTimers[tenantID] = timer
	m.mu.Unlock()
}

// KickWarmStorageTierScan is the poller's durable rediscovery hook. Resolve
// the tenant in the central DB, then probe the existing pool cache before
// enqueueing work; a cold tenant is left for a future natural warm kick.
func (m *tenantWorkerManager) KickWarmStorageTierScan(tenantID string) {
	if m == nil || m.meta == nil || m.pool == nil || tenantID == "" {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	tenantRecord, err := m.meta.GetTenant(ctx, tenantID)
	if err != nil || tenantRecord.Status != meta.TenantActive {
		return
	}
	_, release, ok := m.pool.AcquireCached(tenantRecord)
	if !ok {
		return
	}
	release()
	m.KickWithOrg(tenantID, tenantRecord.TiDBCloudOrgID, WorkStorageTierScan)
}

// logFileGCReKick emits the tenant-attributed FileGC re-kick signal at the
// default-visible Info level, rate-limited per tenant so a sustained backlog
// drain cannot flood logs. No-op when reKick has no WorkFileGC bit.
func (m *tenantWorkerManager) logFileGCReKick(ctx context.Context, tenantID, orgID string, reKick, drainCapMask, pendingMask int) {
	if reKick&WorkFileGC == 0 {
		return
	}
	if !m.shouldRateLimitedLog(tenantID, "rekick_file_gc", tenantWorkerSlotBusyLogInterval) {
		return
	}
	logger.Info(ctx, "tenant_worker_rekick",
		zap.String("tenant_id", tenantID),
		zap.Int("work_mask", reKick),
		zap.Int("drain_cap_mask", drainCapMask),
		zap.Int("pending_mask", pendingMask),
		zap.String("tidbcloud_org_id", orgID))
}

// observeFileGCClaimableBacklog samples claimable queue depth / oldest age for
// sparse per-tenant gauges. Best-effort: errors are ignored so observation
// never blocks scheduling.
func (m *tenantWorkerManager) observeFileGCClaimableBacklog(ctx context.Context, target *tenantTarget) {
	if target == nil || target.store == nil || ctx.Err() != nil {
		return
	}
	if !m.ownsFileGC(target.tenantID) {
		metrics.ClearFileGCClaimableBacklog(target.tenantID)
		return
	}
	now := time.Now().UTC()
	n, oldest, err := target.store.ObserveClaimableFileGCTasks(ctx, now)
	if err != nil {
		return
	}
	var age time.Duration
	if n > 0 && !oldest.IsZero() {
		age = now.Sub(oldest)
	}
	orgID := target.metricOrgID()
	m.withOwnedFileGCGauges(target.tenantID, func() {
		metrics.ObserveFileGCClaimableBacklog(target.tenantID, orgID, n, age)
	})
}

// drainFileGC recovers expired file_gc leases and drains available tasks.
// Returns true if the drain hit its batch cap (more work likely remains).
func (m *tenantWorkerManager) drainFileGC(ctx context.Context, target *tenantTarget) (hitCap bool) {
	if ctx.Err() != nil {
		return false
	}
	if target == nil {
		return false
	}
	if !m.ownsFileGC(target.tenantID) {
		metrics.ClearFileGCClaimableBacklog(target.tenantID)
		return false
	}
	b := target.backend
	if b == nil {
		return false
	}
	orgID := target.metricOrgID()
	metrics.TouchFileGCDrainHitCap(target.tenantID, orgID)
	now := time.Now().UTC()
	if recovered, err := target.store.RecoverExpiredFileGCTasks(ctx, now, fileGCRecoverLimit); err != nil {
		if !isContextDoneErr(err) {
			logger.Warn(ctx, "tenant_worker_file_gc_recover_failed",
				zap.String("tenant_id", target.tenantID), zap.Error(err))
			metrics.RecordTenantOperationCountWithOrg(target.tenantID, orgID, "file_gc", "recover", "error")
		}
	} else if recovered > 0 {
		// Match recoverExpired: only meter successful recover when work happened.
		metrics.RecordTenantOperationCountWithOrg(target.tenantID, orgID, "file_gc", "recover", "ok")
	}
	// Claim empty + COUNT>0 is not "claimable for alert": SKIP LOCKED is the
	// progress oracle. Keep empty_with_backlog as a counter and clear the
	// stale-backlog gauges so leftover COUNT cannot page.
	claimMismatch := false
	claimStalled := false
	defer func() {
		m.withOwnedFileGCGauges(target.tenantID, func() {
			if claimMismatch {
				// Skip COUNT here: a later ObserveClaimable can still see due rows
				// that SKIP LOCKED could not claim, which is what paged leftovers.
				metrics.ObserveFileGCClaimableBacklog(target.tenantID, orgID, 0, 0)
				metrics.ObserveFileGCDrainHitCap(target.tenantID, orgID, false)
				metrics.ObserveFileGCClaimStalled(target.tenantID, orgID, claimStalled)
				return
			}
			metrics.ObserveFileGCClaimStalled(target.tenantID, orgID, false)
			metrics.ObserveFileGCDrainHitCap(target.tenantID, orgID, hitCap)
			m.observeFileGCClaimableBacklog(ctx, target)
			if hitCap {
				m.clearFileGCMismatchBackoff(target.tenantID)
			}
		})
	}()

	processedCount := 0
	for range fileGCDrainBatchSize {
		if ctx.Err() != nil {
			return false
		}
		// Match main semantics: ProcessOne returns (true, err) when a task was
		// claimed and then failed process/ack after a successful RetryFileGCTask
		// (or still returns true when retry itself fails). One transient task
		// failure must not abort the batch — later due tasks still drain in this
		// kick (edge-triggered FileGC has no guaranteed external re-kick).
		// (false, err) is claim/infrastructure failure: stop this kick.
		// (false, nil) is empty claim: stop with backlog diagnostics.
		processed, err := b.ProcessOneFileGCTask(ctx)
		metrics.TouchFileGCDrainHitCap(target.tenantID, orgID)
		if err != nil {
			// Skip log + failure metrics on shutdown/cancel so rolling restarts
			// do not pollute drive9_tenant_operation_failures_total (recover
			// already suppresses context-done metrics the same way).
			if !isContextDoneErr(err) {
				logger.Warn(ctx, "tenant_worker_file_gc_process_failed",
					zap.String("tenant_id", target.tenantID),
					zap.Int("processed_this_kick", processedCount),
					zap.Bool("claimed", processed),
					zap.Error(err))
				if processed {
					// Task left the claimable queue (retry scheduled or stuck
					// processing). Continue the batch; attribute per-task error.
					metrics.RecordTenantOperationCountWithOrg(target.tenantID, orgID, "file_gc", "drain", "task_error")
				} else {
					// No task claimed — further ProcessOne calls likely fail the same way.
					metrics.RecordTenantOperationCountWithOrg(target.tenantID, orgID, "file_gc", "drain", "stop_error")
					return false
				}
			}
		}
		if !processed {
			// Claim empty: either truly idle, or claim cannot see due rows
			// (scope/clock). Count claimable rows (available_at <= now), not all
			// queued — in-backoff retries must not look like a stuck backlog.
			// Use a fresh clock so a row that became due mid-batch is not
			// undercounted relative to ClaimFileGCTask inside ProcessOne.
			// Only run when err == nil (empty queue); claim errors already returned.
			if err != nil {
				return false
			}
			n, _, qerr := m.observeClaimableAfterEmptyClaim(ctx, target)
			if qerr != nil {
				if !isContextDoneErr(qerr) {
					logger.Warn(ctx, "tenant_worker_file_gc_count_claimable_failed",
						zap.String("tenant_id", target.tenantID),
						zap.Error(qerr))
					metrics.RecordTenantOperationCountWithOrg(target.tenantID, orgID, "file_gc", "drain", "stop_empty_count_error")
				}
				// Do not re-export COUNT after a failed observe: that path can
				// republish the same phantom leftover this change exists to drop.
				claimMismatch = true
				if !m.scheduleFileGCMismatchReKick(target) {
					claimStalled = true
				}
			} else if n > 0 {
				if m.shouldRateLimitedLog(target.tenantID, "claim_empty_with_backlog", tenantWorkerSlotBusyLogInterval) {
					logger.Warn(ctx, "tenant_worker_file_gc_claim_empty_with_backlog",
						zap.String("tenant_id", target.tenantID),
						zap.Int64("claimable_queued", n),
						zap.Int("processed_this_kick", processedCount),
						zap.String("tidbcloud_org_id", orgID))
				}
				// Neutral aggregate signal (not a failure result — no tenant_id series).
				// Keep operation "drain" so all drain outcomes share one dimension.
				metrics.RecordTenantOperationCountWithOrg(target.tenantID, orgID, "file_gc", "drain", "empty_with_backlog")
				claimMismatch = true
				if !m.scheduleFileGCMismatchReKick(target) {
					claimStalled = true
				}
			} else {
				metrics.RecordTenantOperationCountWithOrg(target.tenantID, orgID, "file_gc", "drain", "stop_empty")
				m.clearFileGCMismatchBackoff(target.tenantID)
				// Clear via claimMismatch so a failed deferred COUNT cannot
				// leave the previous kick's claimable series exported.
				claimMismatch = true
			}
			return false
		}
		processedCount++
	}
	metrics.RecordTenantOperationCountWithOrg(target.tenantID, orgID, "file_gc", "drain", "hit_cap")
	return true // drained the full batch — likely more remains
}

func (m *tenantWorkerManager) observeClaimableAfterEmptyClaim(ctx context.Context, target *tenantTarget) (int64, time.Time, error) {
	if drainFileGCClaimableAfterEmptyTestHook != nil {
		return drainFileGCClaimableAfterEmptyTestHook()
	}
	return target.store.ObserveClaimableFileGCTasks(ctx, time.Now().UTC())
}

func (m *tenantWorkerManager) scheduleFileGCMismatchReKick(target *tenantTarget) bool {
	if m == nil || target == nil || m.kicks == nil {
		return false
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.stopped {
		return false
	}
	if m.fileGCMismatchTimers == nil {
		m.fileGCMismatchTimers = make(map[string]*time.Timer)
	}
	if m.fileGCMismatchAttempts == nil {
		m.fileGCMismatchAttempts = make(map[string]int)
	}
	if _, armed := m.fileGCMismatchTimers[target.tenantID]; armed {
		return true
	}
	attempt := m.fileGCMismatchAttempts[target.tenantID]
	if attempt >= fileGCMismatchReKickMaxAttempts {
		return false
	}
	delay := fileGCMismatchReKickBase << attempt
	if delay > fileGCMismatchReKickMax || delay <= 0 {
		delay = fileGCMismatchReKickMax
	}
	m.fileGCMismatchAttempts[target.tenantID] = attempt + 1
	tenantID, orgID := target.tenantID, target.metricOrgID()
	var timer *time.Timer
	timer = time.AfterFunc(delay, func() {
		m.mu.Lock()
		if cur, ok := m.fileGCMismatchTimers[tenantID]; ok && cur == timer {
			delete(m.fileGCMismatchTimers, tenantID)
		}
		stopped := m.stopped
		m.mu.Unlock()
		if stopped {
			return
		}
		if !m.ownsFileGC(tenantID) {
			metrics.ClearFileGCClaimableBacklog(tenantID)
			m.clearFileGCMismatchBackoff(tenantID)
			return
		}
		m.KickWithOrg(tenantID, orgID, WorkFileGC)
	})
	m.fileGCMismatchTimers[target.tenantID] = timer
	return true
}

func (m *tenantWorkerManager) clearFileGCMismatchBackoff(tenantID string) {
	if m == nil || tenantID == "" {
		return
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	if t, ok := m.fileGCMismatchTimers[tenantID]; ok {
		t.Stop()
		delete(m.fileGCMismatchTimers, tenantID)
	}
	delete(m.fileGCMismatchAttempts, tenantID)
}

// recoverExpired recovers expired semantic task leases for the tenant.
func (m *tenantWorkerManager) recoverExpired(ctx context.Context, target *tenantTarget) {
	if ctx.Err() != nil {
		return
	}
	start := time.Now()
	recovered, err := target.store.RecoverExpiredSemanticTasks(ctx, time.Now().UTC(), semanticRecoverLimit)
	if err != nil {
		if !isContextDoneErr(err) {
			metrics.RecordTenantOperationWithOrg(target.tenantID, target.metricOrgID(), "semantic_worker", "recover", "error", time.Since(start))
			logger.Warn(ctx, "tenant_worker_recover_failed",
				zap.String("tenant_id", target.tenantID), zap.Error(err))
		}
		return
	}
	if recovered > 0 {
		metrics.RecordTenantOperationWithOrg(target.tenantID, target.metricOrgID(), "semantic_worker", "recover", "ok", time.Since(start))
		logger.Info(ctx, "tenant_worker_recover_ok",
			zap.String("tenant_id", target.tenantID),
			zap.Int("recovered", recovered))
	}
}

// piggybackMaintenance runs fs_events cleanup and file-count observation for
// the tenant, throttled to once per MaintenanceInterval. This replaces the old
// independent cleanupFSEvents leader goroutine and collectObservation scan.
// Semantic queue observation is intentionally NOT part of this throttle: it
// runs on its own cadence (observeTenantThrottled / observeTenantNow) so lag
// alerting stays fresh when MaintenanceInterval is stretched.
func (m *tenantWorkerManager) piggybackMaintenance(ctx context.Context, target *tenantTarget) {
	if ctx.Err() != nil {
		return
	}
	now := time.Now()
	m.mu.Lock()
	last := m.lastMaintenance[target.tenantID]
	if now.Sub(last) < m.opts.MaintenanceInterval {
		m.mu.Unlock()
		return
	}
	m.lastMaintenance[target.tenantID] = now
	m.mu.Unlock()

	// fs_events cleanup: prune rows older than the configured retention.
	// Deletes are batched (fsEventsSweepBatchSize rows per statement, capped at
	// fsEventsSweepMaxBatches per sweep) so a hot tenant's first sweep after a
	// long over-retention period cannot hit TiDB transaction size limits.
	// hasMore means the cap was hit with leftover rows; they drain on the next
	// maintenance cycle (or via the lazy write-path sweep).
	if count, err := target.store.CountFSEvents(ctx); err != nil {
		metrics.RecordTenantOperationWithOrg(target.tenantID, target.metricOrgID(), "event_bus", "fs_events_count", metrics.ResultForError(err), 0)
	} else if count > 0 {
		metrics.RecordFSEventsRowsWithOrg(target.tenantID, target.metricOrgID(), count)
	} else {
		metrics.DeleteFSEventsRowsWithOrg(target.tenantID, target.metricOrgID())
	}
	if target.store.Scope().Shared() {
		// Shared schema: every tenant lives in one physical fs_events table,
		// so dead/idle tenants' rows are unreachable by any fs_id-scoped
		// sweep. Run the pool-wide sweep via the server-level helper instead
		// (detached, cluster-throttled; wired in NewWithConfig). Running here
		// (piggybackMaintenance only fires for already-warm tenants) is the
		// design's "no cold wakeup" gate: maintenance of an active shared
		// resource, not a cron over cold DBs.
		if m.opts.SweepSharedFSEvents != nil {
			m.opts.SweepSharedFSEvents(target.store, target.metricOrgID())
		}
	} else if n, hasMore, err := target.store.DeleteFSEventsBefore(ctx, now.Add(-m.opts.FSEventsRetention), fsEventsSweepBatchSize, fsEventsSweepMaxBatches); err != nil {
		metrics.RecordTenantOperationWithOrg(target.tenantID, target.metricOrgID(), "event_bus", "retention_sweep", metrics.ResultForError(err), 0)
		if ctx.Err() == nil {
			logger.Warn(ctx, "tenant_worker_fs_events_cleanup_failed",
				zap.String("tenant_id", target.tenantID), zap.Error(err))
		}
	} else {
		metrics.RecordTenantOperationWithOrg(target.tenantID, target.metricOrgID(), "event_bus", "retention_sweep", "ok", 0)
		metrics.RecordFSEventsPruned(n)
		if hasMore {
			logger.Info(ctx, "tenant_worker_fs_events_cleanup_has_more",
				zap.String("tenant_id", target.tenantID),
				zap.Int64("deleted", n))
		}
	}

	// Delete-orphan reconciliation (§4.2.1): one throttled round per tenant in
	// repair mode. The scan cursor is held per tenant in this manager
	// (in-memory): a pod restart or shard owner change re-covers earlier
	// rows — repairs persist, so this only delays tail coverage for very
	// large tenants, it never loses repairs.
	m.reconcileDeleteOrphans(ctx, target)

	// File-count observation: refresh the per-tenant file-count gauge (with
	// the db9/s3 storage-tier split once the tier scan is ready) at
	// maintenance cadence. The per-tier COUNT reads stay off request hot
	// paths by design. Publishing is shard-owner fenced so a ring move cannot
	// leave a stale value exported on a non-owner pod.
	if target.backend != nil {
		m.withOwnedFileCountGauge(target.tenantID, func() {
			target.backend.ObserveTenantFileCounts(ctx)
		})
	}
}

// withOwnedFileCountGauge owner-fences drive9_tenant_file_count publishes on
// this process. Non-owners delete leftover series instead of publishing; a
// ring change between the two ownership checks also drops the just-written
// series, mirroring withOwnedFileGCGauges.
func (m *tenantWorkerManager) withOwnedFileCountGauge(tenantID string, publish func()) {
	if m == nil || tenantID == "" || publish == nil {
		return
	}
	if !m.ownsShard(tenantID) {
		metrics.ClearTenantFileCount(tenantID)
		return
	}
	defer func() {
		if !m.ownsShard(tenantID) {
			metrics.ClearTenantFileCount(tenantID)
		}
	}()
	publish()
}

// dropUnownedFileCountGauges deletes this process's drive9_tenant_file_count
// series for tenants the current shard ring no longer assigns here. Per-tenant
// retention cannot expire them while this pod still serves the tenant's
// requests (any tenant-attributed write refreshes the tenant's lastSeen), so
// without this a stale value would keep winning cross-pod max aggregations
// against the new owner's series.
func (m *tenantWorkerManager) dropUnownedFileCountGauges() {
	if m == nil {
		return
	}
	dropped := 0
	for _, tenantID := range metrics.TenantFileCountTenantIDs() {
		if m.ownsShard(tenantID) {
			continue
		}
		metrics.ClearTenantFileCount(tenantID)
		dropped++
	}
	if dropped > 0 {
		logger.Info(context.Background(), "tenant_file_count_gauges_dropped_unowned",
			zap.Int("tenants", dropped))
	}
}

// dropUnownedShardGauges drops this process's shard-owned observation gauges
// (FileGC backlog, tenant file count) for tenants the ring moved elsewhere.
func (m *tenantWorkerManager) dropUnownedShardGauges() {
	m.dropUnownedFileGCGauges()
	m.dropUnownedFileCountGauges()
}

// requestDeleteReconcile is the consumer side of WorkDeleteReconcile. When
// the per-tenant reconcile throttle has elapsed it runs the sweep now;
// otherwise it arms exactly one delayed re-kick so the repair still runs once
// the throttle window opens — even if the tenant goes completely idle after
// the delete (gate 4: the residual-window backstop must not depend on
// unrelated future work).
func (m *tenantWorkerManager) requestDeleteReconcile(ctx context.Context, target *tenantTarget) {
	if ctx.Err() != nil {
		return
	}
	now := time.Now()
	m.mu.Lock()
	if m.lastReconcile == nil {
		m.lastReconcile = make(map[string]time.Time)
	}
	last := m.lastReconcile[target.tenantID]
	if now.Sub(last) >= tenantDeleteReconcileInterval {
		m.mu.Unlock()
		m.runDeleteReconcile(ctx, target)
		return
	}
	// Throttled: arm one delayed re-kick for when the window opens. Multiple
	// kicks in the window share the single armed timer (no re-kick storm).
	remaining := tenantDeleteReconcileInterval - now.Sub(last)
	if remaining < 0 {
		remaining = 0
	}
	m.armDeleteReconcileReKick(target, remaining)
	m.mu.Unlock()
	// Durable counterpart of the in-process timer: the pending row survives a
	// pod restart (the outbox poller re-dispatches due rows), so a throttled
	// reconcile is never lost with the process. Only extend an EXISTING row
	// (atomic UPDATE): a stale duplicate kick that arrives after a successful
	// sweep must not re-create a pending marker for work already done (the
	// original signal always leaves a row via dispatch/failure, so skipping
	// here loses nothing).
	m.extendDeleteReconcilePending(ctx, target.tenantID, now.Add(remaining))
}

// armDeleteReconcileReKick schedules exactly one delayed WorkDeleteReconcile
// re-kick for the tenant (shared by all requests inside the same throttle
// window; the timer is removed when it fires or when the sweep runs). The
// caller must hold m.mu.
func (m *tenantWorkerManager) armDeleteReconcileReKick(target *tenantTarget, delay time.Duration) {
	if m.reconcileTimers == nil {
		m.reconcileTimers = make(map[string]*time.Timer)
	}
	if _, ok := m.reconcileTimers[target.tenantID]; ok {
		return
	}
	if delay < 0 {
		delay = 0
	}
	tenantID, orgID := target.tenantID, target.metricOrgID()
	var timer *time.Timer
	timer = time.AfterFunc(delay, func() {
		m.mu.Lock()
		// Only remove OUR entry: a stale callback racing a concurrent
		// re-arm must not delete the newly armed timer's map slot (the new
		// timer still fires, but would become uncancellable).
		if cur, ok := m.reconcileTimers[tenantID]; ok && cur == timer {
			delete(m.reconcileTimers, tenantID)
		}
		m.mu.Unlock()
		m.KickWithOrg(tenantID, orgID, WorkDeleteReconcile)
	})
	m.reconcileTimers[target.tenantID] = timer
}

// reconcileDeleteOrphans is the piggyback-maintenance entry point: it runs
// one delete-orphan reconciliation round when the per-tenant throttle has
// elapsed. Like the rest of piggyback maintenance it only fires for
// already-warm (kick-driven) tenants — it never wakes a cold tenant.
//
// The periodic round is the BACKSTOP, not the primary trigger: deletes
// request a sweep directly via WorkDeleteReconcile and durably record a
// delete_reconcile_pending marker. A full-table scan for a tenant with no
// outstanding delete is pure cost (production fleets showed it finding zero
// orphans for weeks while dominating slow-query volume), so with a meta
// store available the piggyback round runs only when a due marker exists.
// A future-dated marker (an in-flight delete's crash-recovery protection)
// does not qualify: the poller re-dispatches it when it comes due. Without
// a meta store (single-tenant fallback mode) the marker machinery does not
// exist and the round keeps the legacy unconditional throttle behavior.
func (m *tenantWorkerManager) reconcileDeleteOrphans(ctx context.Context, target *tenantTarget) {
	if ctx.Err() != nil {
		return
	}
	m.mu.Lock()
	last := m.lastReconcile[target.tenantID]
	due := time.Since(last) >= tenantDeleteReconcileInterval
	m.mu.Unlock()
	if !due {
		return
	}
	if m.meta != nil {
		dueAt, found, err := m.meta.GetDeleteReconcilePending(ctx, target.tenantID)
		if err != nil {
			if !isContextDoneErr(err) {
				logger.Warn(ctx, "tenant_worker_delete_reconcile_gate_failed",
					zap.String("tenant_id", target.tenantID), zap.Error(err))
			}
			return
		}
		if !found || dueAt.After(time.Now()) {
			return
		}
	}
	m.runDeleteReconcile(ctx, target)
}

// runDeleteReconcile executes one delete-orphan reconciliation round for the
// tenant in repair mode. The throttle timestamp is recorded only on success;
// on failure the pending re-kick timer is re-armed (instead of being left
// cancelled), so an idle tenant still retries instead of waiting for
// unrelated future work.
func (m *tenantWorkerManager) runDeleteReconcile(ctx context.Context, target *tenantTarget) {
	if ctx.Err() != nil {
		return
	}
	sweepStart := time.Now()
	m.mu.Lock()
	if m.lastReconcile == nil {
		m.lastReconcile = make(map[string]time.Time)
	}
	if m.reconcileState == nil {
		m.reconcileState = make(map[string]*datastore.DeleteReconcileState)
	}
	// Stop any pending re-kick while this sweep runs (it would otherwise
	// duplicate work); a failure path below re-arms it.
	if t, ok := m.reconcileTimers[target.tenantID]; ok {
		t.Stop()
		delete(m.reconcileTimers, target.tenantID)
	}
	state := m.reconcileState[target.tenantID]
	if state == nil {
		state = &datastore.DeleteReconcileState{}
		m.reconcileState[target.tenantID] = state
	}
	m.mu.Unlock()

	dryRun := false // repair mode: this sweep is the only backstop for the
	// residual race window, so it must actually fix orphans by default.
	report, err := target.store.ReconcileDeleteOrphans(ctx, dryRun, 0, state)
	if err != nil {
		if !isContextDoneErr(err) {
			logger.Warn(ctx, "tenant_worker_delete_reconcile_failed",
				zap.String("tenant_id", target.tenantID),
				zap.Bool("dry_run", dryRun),
				zap.Error(err))
		}
		m.mu.Lock()
		m.armDeleteReconcileReKick(target, tenantDeleteReconcileInterval)
		m.mu.Unlock()
		// Durable retry: the pending row makes the failed request survive pod
		// restarts, re-dispatched by the outbox poller's pending scan.
		m.persistDeleteReconcilePending(ctx, target.tenantID, time.Now().Add(tenantDeleteReconcileInterval))
		return
	}
	m.mu.Lock()
	m.lastReconcile[target.tenantID] = time.Now()
	m.mu.Unlock()
	if report.Repaired > 0 || state.DentryCursor != "" || state.InodeCursor != "" {
		// The audit is not finished: repairs can strand rows one level down
		// (multi-round convergence), and a scan-capped round leaves an
		// unaudited tail (state cursors not reset). Schedule the follow-up
		// round with ONE atomic marker write — an owed marker (due no later
		// than the sweep start) moves to the follow-up time, a newer
		// delete's marker is untouched, a missing marker is inserted. With
		// the piggyback sweep gated on the marker, a clear-then-insert
		// sequence could lose the follow-up round to a crash or transient
		// meta error between the two writes.
		m.mu.Lock()
		m.armDeleteReconcileReKick(target, tenantDeleteReconcileInterval)
		m.mu.Unlock()
		m.rearmDeleteReconcilePending(ctx, target.tenantID, time.Now().Add(tenantDeleteReconcileInterval), sweepStart)
	} else {
		// Full clean pass: clear only markers that predate this sweep — a
		// pending row written by a NEWER delete while the sweep was running
		// must survive to be re-kicked by the poller scan (its own
		// in-process timer is the only other retry).
		m.clearDeleteReconcilePendingBefore(ctx, target.tenantID, sweepStart)
	}
	if report.BrokenDentries > 0 || report.StrandedInodes > 0 {
		logger.Info(ctx, "tenant_worker_delete_reconcile_candidates",
			zap.String("tenant_id", target.tenantID),
			zap.Bool("dry_run", dryRun),
			zap.Int64("broken_dentries", report.BrokenDentries),
			zap.Int64("stranded_inodes", report.StrandedInodes),
			zap.Int64("repaired", report.Repaired))
	}
}

// persistDeleteReconcilePending durably records that the tenant's reconcile
// sweep is owed at dueAt. Best-effort and meta-only: without a meta store
// (tests, minimal fallback setups) the in-process timer remains the retry
// path.
func (m *tenantWorkerManager) persistDeleteReconcilePending(ctx context.Context, tenantID string, dueAt time.Time) {
	if m == nil || m.meta == nil {
		return
	}
	if err := m.meta.UpsertDeleteReconcilePending(ctx, tenantID, dueAt); err != nil {
		if !isContextDoneErr(err) {
			logger.Warn(ctx, "tenant_worker_delete_reconcile_persist_failed",
				zap.String("tenant_id", tenantID), zap.Error(err))
		}
	}
}

// rearmDeleteReconcilePending durably schedules a follow-up reconcile round
// at dueAt after a sweep that repaired rows or stopped mid-pass; cutoff is
// the sweep's start time. One atomic upsert: an owed marker (due <= cutoff)
// moves to dueAt, a newer delete's marker keeps its own due time, a missing
// marker is inserted. Failure is Error-level: with the piggyback sweep
// marker-gated, a lost re-arm silently disables the backstop until the
// tenant's next unrelated delete.
func (m *tenantWorkerManager) rearmDeleteReconcilePending(ctx context.Context, tenantID string, dueAt, cutoff time.Time) {
	if m == nil || m.meta == nil {
		return
	}
	if err := m.meta.RearmDeleteReconcilePending(ctx, tenantID, dueAt, cutoff); err != nil {
		if !isContextDoneErr(err) {
			logger.Error(ctx, "tenant_worker_delete_reconcile_rearm_failed",
				zap.String("tenant_id", tenantID), zap.Error(err))
		}
	}
}

// extendDeleteReconcilePending atomically moves an existing pending row's due
// time forward (throttle extension). A missing row is a no-op: a stale
// duplicate kick after a successful sweep must not re-create a marker.
func (m *tenantWorkerManager) extendDeleteReconcilePending(ctx context.Context, tenantID string, dueAt time.Time) {
	if m == nil || m.meta == nil {
		return
	}
	if _, err := m.meta.ExtendDeleteReconcilePending(ctx, tenantID, dueAt); err != nil {
		if !isContextDoneErr(err) {
			logger.Warn(ctx, "tenant_worker_delete_reconcile_extend_failed",
				zap.String("tenant_id", tenantID), zap.Error(err))
		}
	}
}

// clearDeleteReconcilePending removes the tenant's durable pending marker
// after a successful sweep. Best-effort; a missing row is a no-op.
func (m *tenantWorkerManager) clearDeleteReconcilePending(ctx context.Context, tenantID string) {
	if m == nil || m.meta == nil {
		return
	}
	if err := m.meta.DeleteDeleteReconcilePending(ctx, tenantID); err != nil {
		if !isContextDoneErr(err) {
			logger.Warn(ctx, "tenant_worker_delete_reconcile_clear_failed",
				zap.String("tenant_id", tenantID), zap.Error(err))
		}
	}
}

// clearDeleteReconcilePendingBefore removes the tenant's durable pending
// marker only when it was due no later than cutoff. Best-effort; used by a
// successful sweep with its own start time as the cutoff.
func (m *tenantWorkerManager) clearDeleteReconcilePendingBefore(ctx context.Context, tenantID string, cutoff time.Time) {
	if m == nil || m.meta == nil {
		return
	}
	if err := m.meta.DeleteDeleteReconcilePendingBefore(ctx, tenantID, cutoff); err != nil {
		if !isContextDoneErr(err) {
			logger.Warn(ctx, "tenant_worker_delete_reconcile_clear_before_failed",
				zap.String("tenant_id", tenantID), zap.Error(err))
		}
	}
}

// kickRef resolves a kicked tenant ID to a schedulable ref, applying the same
// status and provider/task-type filters as the scan path. The returned error
// is non-nil only when the tenant could not be resolved due to a transient
// lookup failure; callers must not treat a definitive rejection
// (not found / non-active / no work types) as transient.
func (m *tenantWorkerManager) kickRef(ctx context.Context, tenantID string, workMask int) (semanticTenantRef, bool, error) {
	if tenantID == tenantLocalID {
		if m.meta == nil || m.pool == nil {
			// File GC and delete-reconcile work are viable for ANY fallback
			// backend, even one with no semantic/extract task types: the
			// recursive delete path emits WorkDeleteReconcile unconditionally
			// and must not be dropped here.
			if m.shouldIncludeFallback() || workMask&(WorkFileGC|WorkDeleteReconcile) != 0 {
				return semanticTenantRef{id: tenantLocalID}, true, nil
			}
		}
		return semanticTenantRef{}, false, nil
	}
	if m.meta == nil || m.pool == nil {
		return semanticTenantRef{}, false, nil
	}
	t, err := m.meta.GetTenant(ctx, tenantID)
	if err != nil {
		logger.Warn(ctx, "tenant_worker_kick_tenant_lookup_failed",
			zap.String("tenant_id", tenantID),
			zap.Error(err))
		return semanticTenantRef{}, false, err
	}
	if t.Status != meta.TenantActive {
		return semanticTenantRef{}, false, nil
	}
	if !hasAnyTaskTypes(m.taskTypesForProvider(t.Provider)) && !m.hasShardedWorkForTenant(t.Provider) {
		return semanticTenantRef{}, false, nil
	}
	return semanticTenantRef{id: t.ID, tenant: t}, true, nil
}

// hasShardedWorkForTenant reports whether the tenant provider supports any
// sharded work type (file_gc or quota), independent of semantic task types.
// File_gc and quota outbox processing don't depend on semantic task routing.
func (m *tenantWorkerManager) hasShardedWorkForTenant(provider string) bool {
	// File GC and quota outbox run for all active tenants with a backend.
	// The backend must exist (Acquire succeeds) — we don't need task-type
	// routing to enable them. So any active tenant is eligible.
	return true
}

func (m *tenantWorkerManager) targetForRef(ctx context.Context, ref semanticTenantRef, tidbCloudOrgID string) (*tenantTarget, error) {
	if ref.id == tenantLocalID {
		if m.fallback == nil {
			return nil, fmt.Errorf("backend missing for %s", ref.id)
		}
		return &tenantTarget{
			tenantID:         ref.id,
			tidbCloudOrgID:   m.fallback.TiDBCloudOrgID(),
			backend:          m.fallback,
			store:            m.fallback.Store(),
			allowedTaskTypes: m.taskTypesForTarget(m.fallback),
			release:          func() {},
		}, nil
	}
	if ref.tenant == nil {
		return nil, fmt.Errorf("tenant metadata missing for %s", ref.id)
	}
	acquireStart := time.Now()
	b, release, err := m.pool.Acquire(ctx, ref.tenant)
	if err != nil {
		// Kick-driven acquire failure: a kick arrived but the tenant TiDB could
		// not be opened. Record so the major alert can detect sustained worker
		// acquire errors (kicks not reaching the tenant DB).
		metrics.RecordTenantOperationWithOrg(ref.tenant.ID, tidbCloudOrgID, "user_db_access", "tenant_worker_acquire", metrics.ResultForError(err), time.Since(acquireStart))
		return nil, fmt.Errorf("acquire tenant backend: %w", err)
	}
	if b == nil {
		release()
		metrics.RecordTenantOperationWithOrg(ref.tenant.ID, tidbCloudOrgID, "user_db_access", "tenant_worker_acquire", "error", time.Since(acquireStart))
		return nil, fmt.Errorf("backend missing for %s", ref.id)
	}
	// Kick-driven acquire success: the tenant TiDB is now open for this kick.
	// The rate follows write traffic (kicks are produced by writes). A spike
	// uncorrelated with writes would suggest a scan path regressing.
	metrics.RecordTenantOperationWithOrg(ref.tenant.ID, b.TiDBCloudOrgID(), "user_db_access", "tenant_worker_acquire", "ok", time.Since(acquireStart))
	return &tenantTarget{
		tenantID:         ref.id,
		tidbCloudOrgID:   b.TiDBCloudOrgID(),
		backend:          b,
		store:            b.Store(),
		allowedTaskTypes: m.taskTypesForTarget(b),
		release:          release,
	}, nil
}

func (m *tenantWorkerManager) invalidateTenantBackend(tenantID string) {
	if tenantID == tenantLocalID {
		return
	}
	if m.pool == nil {
		return
	}
	m.pool.Invalidate(tenantID)
}

// setExtractRuntimeDeps wires the execution-time extract config resolver and
// the cluster image-extract kill switch (shared with the Server) into the
// worker. Either may be nil: a nil resolver selects the process-default
// backend runtime, a nil switch means image extract is allowed.
func (m *tenantWorkerManager) setExtractRuntimeDeps(resolver *extractConfigResolver, globalDisabled *atomic.Bool) {
	m.extractResolver = resolver
	m.imageExtractGlobalDisabled = globalDisabled
}

// imageExtractGloballyDisabled reports whether the cluster image-extract kill
// switch is on.
func (m *tenantWorkerManager) imageExtractGloballyDisabled() bool {
	return m != nil && m.imageExtractGlobalDisabled != nil && m.imageExtractGlobalDisabled.Load()
}

// claimableTaskTypes returns the target's allowed semantic task types with
// img_extract_text filtered out while the cluster kill switch is on. The
// switch stops claiming NEW image tasks only; file writes still enqueue them.
func (m *tenantWorkerManager) claimableTaskTypes(target *tenantTarget) []semantic.TaskType {
	types := target.allowedTaskTypes
	if !m.imageExtractGloballyDisabled() {
		return types
	}
	filtered := make([]semantic.TaskType, 0, len(types))
	for _, t := range types {
		if t == semantic.TaskTypeImgExtractText {
			continue
		}
		filtered = append(filtered, t)
	}
	return filtered
}

// claimAndProcessOne claims one semantic task and processes it. claimed is
// true if a task was claimed (and the caller should try again). queueEmpty is
// true only when the claim found no claimable task — never on claim errors or
// when no task type is claimable, so callers can safely use it as a
// "queue just drained" signal.
func (m *tenantWorkerManager) claimAndProcessOne(ctx context.Context, target *tenantTarget) (claimed, queueEmpty bool) {
	claimStart := time.Now()
	allowedTypes := m.claimableTaskTypes(target)
	if len(allowedTypes) == 0 {
		return false, false
	}
	task, found, err := target.store.ClaimSemanticTask(ctx, time.Now().UTC(), m.opts.LeaseDuration, allowedTypes...)
	if err != nil {
		metrics.RecordTenantOperationWithOrg(target.tenantID, target.metricOrgID(), "semantic_worker", "claim", "error", time.Since(claimStart))
		logger.Warn(ctx, "tenant_worker_claim_failed",
			append([]zap.Field{
				zap.String("tenant_id", target.tenantID),
				zap.String("result", "error"),
			}, zap.Error(err))...)
		m.invalidateTenantBackend(target.tenantID)
		return false, false
	}
	if !found {
		return false, true
	}
	metrics.RecordTenantOperationWithOrg(target.tenantID, target.metricOrgID(), "semantic_worker", "claim", "ok", time.Since(claimStart))
	logger.Info(ctx, "tenant_worker_claim_ok",
		append([]zap.Field{
			zap.String("tenant_id", target.tenantID),
			zap.String("result", "ok"),
		}, semanticTaskLogFields(task)...)...)
	m.markProcessingStart()
	defer m.markProcessingDone()
	m.processTask(ctx, target, task)
	return true, false
}

// tryClaimTenantSlot attempts to acquire a per-tenant concurrency slot.
func (m *tenantWorkerManager) tryClaimTenantSlot(tenantID string) bool {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.inflight[tenantID] >= m.opts.PerTenantConcurrency {
		return false
	}
	m.inflight[tenantID]++
	return true
}

func (m *tenantWorkerManager) tenantSlotInflight(tenantID string) int {
	m.mu.Lock()
	defer m.mu.Unlock()
	return m.inflight[tenantID]
}

// shouldRateLimitedLog rate-limits Warn logs per tenant and log name.
func (m *tenantWorkerManager) shouldRateLimitedLog(tenantID, name string, interval time.Duration) bool {
	if m == nil || tenantID == "" || name == "" {
		return false
	}
	if interval <= 0 {
		interval = tenantWorkerSlotBusyLogInterval
	}
	key := tenantID + "\x00" + name
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.lastRateLimitedLog == nil {
		m.lastRateLimitedLog = make(map[string]time.Time)
	}
	now := time.Now()
	if last, ok := m.lastRateLimitedLog[key]; ok && now.Sub(last) < interval {
		return false
	}
	m.lastRateLimitedLog[key] = now
	return true
}

func (m *tenantWorkerManager) releaseTenantSlot(tenantID string) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.inflight[tenantID] <= 1 {
		delete(m.inflight, tenantID)
		return
	}
	m.inflight[tenantID]--
}

func (m *tenantWorkerManager) markProcessingStart() {
	m.mu.Lock()
	m.processing++
	inflight := m.processing
	m.mu.Unlock()
	metrics.RecordGauge("semantic_worker", "inflight", float64(inflight))
}

func (m *tenantWorkerManager) markProcessingDone() {
	m.mu.Lock()
	if m.processing > 0 {
		m.processing--
	}
	inflight := m.processing
	m.mu.Unlock()
	metrics.RecordGauge("semantic_worker", "inflight", float64(inflight))
}

func (m *tenantWorkerManager) appManagedTaskTypes() []semantic.TaskType {
	if m == nil {
		return nil
	}
	if m.pool != nil || (m.fallback != nil && !m.fallback.UsesDatabaseAutoEmbedding()) {
		return tenantWorkerAllowedEmbedTaskTypes
	}
	return appManagedTenantTaskTypes(m.embedder)
}

func (m *tenantWorkerManager) poolExtractTaskTypes() []semantic.TaskType {
	if m == nil || m.pool == nil {
		return nil
	}
	return m.pool.AutoSemanticTaskTypes()
}

func (m *tenantWorkerManager) fallbackExtractTaskTypes() []semantic.TaskType {
	if m == nil || m.fallback == nil {
		return nil
	}
	var out []semantic.TaskType
	if m.fallback.SupportsImageExtractTasks() {
		out = append(out, semantic.TaskTypeImgExtractText)
	}
	if m.fallback.SupportsAudioExtractTasks() {
		out = append(out, semantic.TaskTypeAudioExtractText)
	}
	if m.fallback.SupportsVideoExtractTasks() {
		out = append(out, semantic.TaskTypeVideoExtractVisual)
	}
	return out
}

func (m *tenantWorkerManager) taskTypesForProvider(provider string) []semantic.TaskType {
	if m == nil {
		return nil
	}
	// Provider metadata cannot distinguish native auto from native fts_only.
	// Keep embed in this coarse pre-acquire capability set; the resolved
	// backend below is the authoritative mode boundary.
	_ = provider
	return unionTaskTypes(m.appManagedTaskTypes(), m.poolExtractTaskTypes())
}

func (m *tenantWorkerManager) shouldIncludeFallback() bool {
	if m == nil || m.fallback == nil {
		return false
	}
	if m.fallback.UsesDatabaseAutoEmbedding() {
		return hasAnyTaskTypes(m.fallback.AutoSemanticTaskTypes())
	}
	return hasAnyTaskTypes(m.appManagedTaskTypes()) || hasAnyTaskTypes(m.fallbackExtractTaskTypes())
}

func (m *tenantWorkerManager) taskTypesForTarget(b *backend.Dat9Backend) []semantic.TaskType {
	if m == nil || b == nil {
		return nil
	}
	if b.UsesDatabaseAutoEmbedding() {
		return b.AutoSemanticTaskTypes()
	}
	out := []semantic.TaskType{semantic.TaskTypeEmbed, semantic.TaskTypeEmbedBackfill}
	// A resolver-backed worker has capability even when this backend has no
	// process-default provider. Admission stays durable and resolution happens
	// only after claim, so uploads never query MetaDB. Terminal media outcomes
	// discard fresh due work in bounded batches.
	if m.extractResolver != nil {
		out = unionTaskTypes(out, []semantic.TaskType{
			semantic.TaskTypeImgExtractText,
			semantic.TaskTypeAudioExtractText,
			semantic.TaskTypeVideoExtractVisual,
		})
	}
	if b.SupportsImageExtractTasks() {
		out = unionTaskTypes(out, []semantic.TaskType{semantic.TaskTypeImgExtractText})
	}
	if b.SupportsAudioExtractTasks() {
		out = unionTaskTypes(out, []semantic.TaskType{semantic.TaskTypeAudioExtractText})
	}
	if b.SupportsVideoExtractTasks() {
		out = unionTaskTypes(out, []semantic.TaskType{semantic.TaskTypeVideoExtractVisual})
	}
	if len(out) == 0 {
		return nil
	}
	return out
}

// processKickedAfterSlotClaimTestHook is set only by tests. Production keeps it
// nil. When non-nil it runs immediately after tryClaimTenantSlot succeeds
// (slotHeld=true, before target.release owns the slot).
var processKickedAfterSlotClaimTestHook func(tenantID string)

// processKickedAfterStorageTierScanTestHook is set only by tests that need to
// exercise work coalescing in the narrow window after scan dispatch.
var processKickedAfterStorageTierScanTestHook func(tenantID string)

// pollFallbackOnceTestHook is set only by tests. When non-nil it runs at the
// start of pollFallbackOnce (used to inject panics for workerLoop backoff).
// The manager pointer lets tests ignore concurrent managers started by other
// package tests (t.Parallel) that would otherwise trip a global hook.
var pollFallbackOnceTestHook func(m *tenantWorkerManager)

func isContextDoneErr(err error) bool {
	if err == nil {
		return false
	}
	// Prefer errors.Is so wrapped cancels from process+retry / DB layers still
	// suppress drain failure metrics during rolling restarts.
	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
		return true
	}
	if errors.Is(err, sql.ErrConnDone) {
		return true
	}
	// "database is closed" arises from go-sql-driver/mysql when the underlying
	// connection is torn down during shutdown; treat it as a context-done signal
	// so it is not logged as an unexpected error.
	if strings.Contains(err.Error(), "database is closed") {
		return true
	}
	return false
}

// failpoint injection hook used by semantic task processing (preserved from
// semantic_worker.go for failpoint tests).
func (m *tenantWorkerManager) injectBeforeSemanticTaskFinalize(tenantID string, store *datastore.Store, task *semantic.Task, outcome semanticTaskOutcome) {
	failpoint.InjectCall("semanticWorkerBeforeFinalize", tenantID, store, task, string(outcome.action), outcome.message, outcome.result)
}