queue.go

// queue.go — the observable workqueue.
//
// This file wraps client-go's real workqueue.Type (the same queue every
// controller uses) with SHADOW COPIES of its two internal sets — "dirty"
// and "processing" — so we can see and log the decisions the real queue
// makes silently:
//
//	Add(key)  when key is dirty & not processing  → no-op        [DEDUP]
//	Add(key)  when key is processing              → mark dirty,  [REQUEUE-WHILE-PROCESSING]
//	                                                redelivered exactly once at Done()
//	Add(key)  when key is neither                 → enters FIFO
//
// It also inlines what client-go's rateLimitingType.AddRateLimited does
// (RateLimiter.When + time.AfterFunc + Add) — verbatim the same logic, but
// exposed so we can log the computed backoff as [RATELIMIT].
//
// Honest caveat: the shadow sets mirror the real sets from OUTSIDE the real
// queue's lock. Under extreme rates a microsecond race between a worker's
// Get() and a concurrent Add() can swap a [DEDUP]/[REQUEUE-WHILE-PROCESSING]
// label. The per-key delivery sequence numbers (seq=N) are exact — they are
// the source of truth for "redelivered exactly once".
package main

import (
	"fmt"
	"log"
	"sync"
	"sync/atomic"
	"time"

	"k8s.io/client-go/util/workqueue"
)

// watchKey gets full-fidelity logs (except [ENQUEUE]/[DEDUP], which are
// always counter-sampled — at 50k/sec they are pure noise). All other keys
// are sampled 1-in-*logSample. Pass --log-sample=1 to log EVERYTHING.
const watchKey = "default/obj-0"

// stats is the exact accounting behind the [STATS] line. Log lines are
// sampled; these counters are not.
type stats struct {
	enqueued  atomic.Int64 // external Add() calls (generator / informer)
	adds      atomic.Int64 // ALL add() calls, incl. rate-limit timer re-adds
	dedup     atomic.Int64 // Add() calls that were complete no-ops
	rwp       atomic.Int64 // Add() calls that hit a processing key (marked for redelivery)
	processed atomic.Int64 // worker Done() calls
	failures  atomic.Int64 // AddRateLimited() calls
	triggers  atomic.Int64 // /trigger HTTP requests (pipeline mode)

	mu             sync.Mutex
	perKeyFailures map[string]int64 // cumulative, for the top-5 list

	latMu   sync.Mutex
	lat     []time.Duration // ring buffer of /trigger latencies (pipeline mode)
	latIdx  int
	latFull bool
}

func newStats() *stats {
	return &stats{perKeyFailures: map[string]int64{}, lat: make([]time.Duration, 4096)}
}

func (s *stats) recordFailure(key string) {
	s.failures.Add(1)
	s.mu.Lock()
	s.perKeyFailures[key]++
	s.mu.Unlock()
}

func (s *stats) recordLatency(d time.Duration) {
	s.latMu.Lock()
	s.lat[s.latIdx] = d
	s.latIdx = (s.latIdx + 1) % len(s.lat)
	if s.latIdx == 0 {
		s.latFull = true
	}
	s.latMu.Unlock()
}

// observableQueue wraps the real queue. All Adds (generator, informer,
// rate-limit re-adds) flow through here, so the shadow sets see everything.
type observableQueue struct {
	inner   workqueue.Interface // the real client-go Type: dedup + processing semantics
	limiter workqueue.RateLimiter
	st      *stats

	mu         sync.Mutex
	dirty      map[string]bool // shadow of the real dirty set
	processing map[string]bool // shadow of the real processing set
	seq        map[string]int64
}

func newObservableQueue(st *stats) *observableQueue {
	return &observableQueue{
		inner:   workqueue.New(),
		limiter: workqueue.DefaultControllerRateLimiter(), // = MaxOf( ItemExponentialFailureRateLimiter(5ms base, 1000s cap), BucketRateLimiter(10 qps, 100 burst) )
		st:      st,
		dirty:      map[string]bool{},
		processing: map[string]bool{},
		seq:        map[string]int64{},
	}
}

// Add is called by external producers (load generator, informer handlers).
func (q *observableQueue) Add(key, source string) {
	if n := q.st.enqueued.Add(1); n%int64(*logSample) == 0 {
		log.Printf("[ENQUEUE] key=%s (%s)", key, source)
	}
	q.add(key, source)
}

// add does the classification + the real Add. source is only for log text.
func (q *observableQueue) add(key, source string) {
	q.st.adds.Add(1)
	q.mu.Lock()
	switch {
	case q.processing[key] && q.dirty[key]:
		q.st.dedup.Add(1)
		q.sampledLog("DEDUP", key, q.st.dedup.Load(),
			"already marked for redelivery after current processing — Add is a no-op ("+source+")")
	case q.processing[key]:
		q.st.rwp.Add(1)
		q.alwaysLogWatchKey("REQUEUE-WHILE-PROCESSING", key, q.st.rwp.Load(),
			"a worker holds this key — it stays OUT of the FIFO now, and will be redelivered exactly once when Done() runs (%s)", source)
	case q.dirty[key]:
		q.st.dedup.Add(1)
		q.sampledLog("DEDUP", key, q.st.dedup.Load(),
			"already waiting in the FIFO — Add is a no-op ("+source+")")
	default:
		// fresh key: this Add actually grows the queue
	}
	q.dirty[key] = true
	q.inner.Add(key) // the real queue makes the same decision, silently
	q.mu.Unlock()
}

// AddRateLimited is client-go's rateLimitingType.AddRateLimited, inlined so
// the computed backoff is visible. When() both RETURNS the delay and
// increments the key's failure counter.
func (q *observableQueue) AddRateLimited(key string) {
	delay := q.limiter.When(key)
	attempt := q.limiter.NumRequeues(key)
	q.st.recordFailure(key)
	q.alwaysLogWatchKey("RATELIMIT", key, q.st.failures.Load(),
		"attempt=%d backoff=%s — re-added when the timer fires (5ms×2^(attempt-1), cap 1000s; a success calls Forget and resets this)", attempt, delay)
	time.AfterFunc(delay, func() { q.add(key, "ratelimit-timer") })
}

func (q *observableQueue) Len() int  { return q.inner.Len() }
func (q *observableQueue) ShutDown() { q.inner.ShutDown() }

// get is the worker side. Blocks until a key is available or shutdown.
func (q *observableQueue) get() (key string, seq int64, shutdown bool) {
	item, sd := q.inner.Get()
	if sd {
		return "", 0, true
	}
	key = item.(string)
	q.mu.Lock()
	delete(q.dirty, key)     // shadow: dirty → processing
	q.processing[key] = true //
	q.seq[key]++
	seq = q.seq[key]
	q.mu.Unlock()
	return key, seq, false
}

// done mirrors the real Done(): leaves the processing set; if the key was
// re-added while we worked, the real queue re-pushes it — exactly once.
func (q *observableQueue) done(key string) {
	q.mu.Lock()
	delete(q.processing, key)
	redeliver := q.dirty[key]
	q.inner.Done(key)
	q.mu.Unlock()
	q.st.processed.Add(1)
	if redeliver && key == watchKey {
		log.Printf("[REDELIVERY] key=%s was re-added while processing → Done() re-queued it (once — no matter how many Adds landed)", key)
	}
}

// --- logging helpers -------------------------------------------------------

// sampledLog logs every *logSample-th occurrence of high-volume categories.
func (q *observableQueue) sampledLog(prefix, key string, counter int64, msg string) {
	if counter%int64(*logSample) == 0 {
		log.Printf("[%s] key=%s %s", prefix, key, msg)
	}
}

// alwaysLogWatchKey logs every event for watchKey, sampled for other keys.
func (q *observableQueue) alwaysLogWatchKey(prefix, key string, counter int64, format string, args ...any) {
	if key == watchKey || counter%int64(*logSample) == 0 {
		log.Printf("[%s] key=%s %s", prefix, key, fmt.Sprintf(format, args...))
	}
}

// --- worker ----------------------------------------------------------------

type worker struct {
	id int
	q  *observableQueue
	st *stats
}

func (w *worker) run() {
	for {
		key, seq, shutdown := w.q.get()
		if shutdown {
			return
		}
		w.logLine(key, "Get seq=%d — this worker now exclusively owns the key", seq)

		start := time.Now()
		time.Sleep(time.Duration(*workMS) * time.Millisecond) // simulate reconcile work
		elapsed := time.Since(start).Round(time.Millisecond / 10)

		if failedRoll() {
			w.q.AddRateLimited(key) // logs [RATELIMIT] with the backoff
			w.logLine(key, "Done seq=%d result=FAIL took=%s → AddRateLimited", seq, elapsed)
		} else {
			w.q.limiter.Forget(key) // textbook: Forget on success → backoff resets
			w.logLine(key, "Done seq=%d result=OK took=%s → Forget (backoff reset)", seq, elapsed)
		}
		w.q.done(key)
	}
}

func (w *worker) logLine(key, format string, args ...any) {
	if key == watchKey || w.st.processed.Load()%int64(*logSample) == 0 {
		log.Printf("[WORKER-%d] key=%s %s", w.id, key, fmt.Sprintf(format, args...))
	}
}