package main

// reconcile.go: a Kubernetes-style controller in one file, standard library only.
// Pieces, in the order an interviewer reads them: API objects, an in-memory API
// server (Store) with optimistic concurrency and watch, an informer-like Cache,
// a rate-limited work queue, the Reconciler, the Controller that wires them,
// a fake kubelet, and a scripted scenario in main that asserts every stage.

import (
	"context"
	"errors"
	"fmt"
	"maps"
	"os"
	"runtime"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

// ---------- API objects (what metav1.ObjectMeta and the CRD types give you) ----------

const finalizerName = "webapp.example.com/cleanup"

// Meta is the part of every object the API server owns: name, version, finalizers, deletion.
type Meta struct {
	Name              string
	ResourceVersion   int64 // bumped on every write; an Update must carry the version it read
	Finalizers        []string
	DeletionTimestamp *time.Time // set by Delete while finalizers remain; the object is gone only when they are cleared
	Annotations       map[string]string
}

func (m *Meta) GetMeta() *Meta { return m }

// Object is what the generic Store needs: metadata access and a deep copy of the concrete type.
type Object[T any] interface {
	GetMeta() *Meta
	DeepCopy() T
}

// WebApp is the custom resource. Replicas and Image are spec; ReadyReplicas is status.
type WebApp struct {
	Meta
	Replicas      int
	Image         string
	ReadyReplicas int
}

// Deployment is the owned child. Owner plays the role of an ownerReference.
type Deployment struct {
	Meta
	Owner         string
	Replicas      int
	Image         string
	ReadyReplicas int // written by the kubelet goroutine, read by the controller
}

func copyMeta(m Meta) Meta {
	m.Finalizers = append([]string(nil), m.Finalizers...)
	m.Annotations = maps.Clone(m.Annotations)
	if m.DeletionTimestamp != nil {
		t := *m.DeletionTimestamp
		m.DeletionTimestamp = &t
	}
	return m
}

func (w *WebApp) DeepCopy() *WebApp {
	c := *w
	c.Meta = copyMeta(w.Meta)
	return &c
}

func (d *Deployment) DeepCopy() *Deployment {
	c := *d
	c.Meta = copyMeta(d.Meta)
	return &c
}

var (
	ErrNotFound      = errors.New("not found")
	ErrAlreadyExists = errors.New("already exists")
	ErrConflict      = errors.New("conflict: object was modified; re-read and apply your change again")
)

// ---------- Store: the API server for one resource, with optimistic concurrency and watch ----------

type EventType string

const (
	Added    EventType = "ADDED"
	Modified EventType = "MODIFIED"
	Deleted  EventType = "DELETED"
)

type Event[T any] struct {
	Type   EventType
	Object T
}

type Store[T Object[T]] struct {
	kind        string
	mu          sync.Mutex
	items       map[string]T
	nextRV      int64
	watchers    map[int]chan Event[T]
	nextWatcher int
	Conflicts   int    // how many Updates were rejected with ErrConflict (read through Stats)
	Dropped     int    // events a slow watcher missed (a real informer would relist)
	BeforeWrite func() // test hook: runs once, before the next Update, like a fake client reactor
}

// SetBeforeWrite installs the one-shot hook under the lock; Update takes and clears it.
func (s *Store[T]) SetBeforeWrite(fn func()) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.BeforeWrite = fn
}

func NewStore[T Object[T]](kind string) *Store[T] {
	return &Store[T]{kind: kind, items: map[string]T{}, watchers: map[int]chan Event[T]{}}
}

// commit stores a copy with a fresh ResourceVersion and tells every watcher. Caller holds mu.
func (s *Store[T]) commit(obj T, typ EventType) T {
	s.nextRV++
	obj.GetMeta().ResourceVersion = s.nextRV
	s.items[obj.GetMeta().Name] = obj
	s.emit(Event[T]{Type: typ, Object: obj.DeepCopy()})
	return obj.DeepCopy()
}

func (s *Store[T]) emit(ev Event[T]) {
	for _, ch := range s.watchers {
		select {
		case ch <- ev:
		default:
			s.Dropped++
		}
	}
}

func (s *Store[T]) Create(obj T) (T, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	var zero T
	if _, ok := s.items[obj.GetMeta().Name]; ok {
		return zero, fmt.Errorf("%s %q: %w", s.kind, obj.GetMeta().Name, ErrAlreadyExists)
	}
	return s.commit(obj.DeepCopy(), Added), nil
}

func (s *Store[T]) Get(name string) (T, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	cur, ok := s.items[name]
	if !ok {
		var zero T
		return zero, fmt.Errorf("%s %q: %w", s.kind, name, ErrNotFound)
	}
	return cur.DeepCopy(), nil
}

// Update is optimistic concurrency: the ResourceVersion on obj must equal the stored one.
// If finalizers are now empty on an object marked for deletion, the object is removed.
func (s *Store[T]) Update(obj T) (T, error) {
	s.mu.Lock()
	hook := s.BeforeWrite
	s.BeforeWrite = nil
	s.mu.Unlock()
	if hook != nil {
		hook() // simulates another client writing between our read and our write
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	var zero T
	name := obj.GetMeta().Name
	cur, ok := s.items[name]
	if !ok {
		return zero, fmt.Errorf("%s %q: %w", s.kind, name, ErrNotFound)
	}
	if cur.GetMeta().ResourceVersion != obj.GetMeta().ResourceVersion {
		s.Conflicts++
		return zero, fmt.Errorf("%s %q (sent rv=%d, stored rv=%d): %w", s.kind, name,
			obj.GetMeta().ResourceVersion, cur.GetMeta().ResourceVersion, ErrConflict)
	}
	c := obj.DeepCopy()
	if c.GetMeta().DeletionTimestamp != nil && len(c.GetMeta().Finalizers) == 0 {
		delete(s.items, name)
		s.emit(Event[T]{Type: Deleted, Object: c})
		return c, nil
	}
	return s.commit(c, Modified), nil
}

// Delete removes the object, or only marks it when finalizers are present (the API server rule).
func (s *Store[T]) Delete(name string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	cur, ok := s.items[name]
	if !ok {
		return fmt.Errorf("%s %q: %w", s.kind, name, ErrNotFound)
	}
	m := cur.GetMeta()
	if len(m.Finalizers) > 0 && m.DeletionTimestamp == nil {
		c := cur.DeepCopy()
		now := time.Now()
		c.GetMeta().DeletionTimestamp = &now
		s.commit(c, Modified)
		return nil
	}
	delete(s.items, name)
	s.emit(Event[T]{Type: Deleted, Object: cur.DeepCopy()})
	return nil
}

// Watch returns a channel that first replays every current object as ADDED (the "list"),
// then streams later changes. stop unregisters the watcher and closes the channel.
func (s *Store[T]) Watch() (<-chan Event[T], func()) {
	s.mu.Lock()
	defer s.mu.Unlock()
	ch := make(chan Event[T], 1024)
	names := make([]string, 0, len(s.items))
	for n := range s.items {
		names = append(names, n)
	}
	sort.Strings(names)
	for _, n := range names {
		ch <- Event[T]{Type: Added, Object: s.items[n].DeepCopy()}
	}
	id := s.nextWatcher
	s.nextWatcher++
	s.watchers[id] = ch
	stop := func() {
		s.mu.Lock()
		defer s.mu.Unlock()
		if _, ok := s.watchers[id]; ok {
			delete(s.watchers, id)
			close(ch)
		}
	}
	return ch, stop
}

// Stats returns the conflict and dropped-event counters under the lock, so a test can read
// them while informers and the kubelet are still writing.
func (s *Store[T]) Stats() (conflicts, dropped int) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.Conflicts, s.Dropped
}

// RetryOnConflict re-runs fn while it fails with ErrConflict. fn must re-read the object
// itself, so each attempt applies the change to the latest version (client-go retry.RetryOnConflict).
func RetryOnConflict(attempts int, fn func() error) error {
	var err error
	for i := 0; i < attempts; i++ {
		if err = fn(); !errors.Is(err, ErrConflict) {
			return err
		}
	}
	return err
}

// ---------- Cache: the informer's local copy (Lister) ----------

type Cache[T Object[T]] struct {
	mu    sync.RWMutex
	items map[string]T
}

func NewCache[T Object[T]]() *Cache[T] { return &Cache[T]{items: map[string]T{}} }

func (c *Cache[T]) Get(name string) (T, bool) {
	c.mu.RLock()
	defer c.mu.RUnlock()
	obj, ok := c.items[name]
	return obj, ok
}

// RunInformer keeps cache current from a watch and calls handle for every event until ctx ends.
func RunInformer[T Object[T]](ctx context.Context, store *Store[T], cache *Cache[T], handle func(T)) {
	ch, stop := store.Watch()
	defer stop()
	for {
		select {
		case <-ctx.Done():
			return
		case ev, ok := <-ch:
			if !ok {
				return
			}
			cache.mu.Lock()
			if ev.Type == Deleted {
				delete(cache.items, ev.Object.GetMeta().Name)
			} else {
				cache.items[ev.Object.GetMeta().Name] = ev.Object
			}
			cache.mu.Unlock()
			handle(ev.Object)
		}
	}
}

// ---------- Work queue: dedup, one worker per key, exponential backoff (client-go workqueue) ----------

type Queue struct {
	mu         sync.Mutex
	cond       *sync.Cond
	order      []string
	dirty      map[string]bool // keys waiting; Add of a waiting key is a no-op
	processing map[string]bool // keys a worker holds; Add while processing re-adds on Done
	failures   map[string]int
	timers     []*time.Timer
	down       bool
	base, max  time.Duration
	Backoffs   []time.Duration // every delay handed out, for the scenario to check
}

func NewQueue(base, max time.Duration) *Queue {
	q := &Queue{dirty: map[string]bool{}, processing: map[string]bool{}, failures: map[string]int{}, base: base, max: max}
	q.cond = sync.NewCond(&q.mu)
	return q
}

func (q *Queue) Add(key string) {
	q.mu.Lock()
	defer q.mu.Unlock()
	if q.down || q.dirty[key] {
		return
	}
	q.dirty[key] = true
	if q.processing[key] {
		return
	}
	q.order = append(q.order, key)
	q.cond.Signal()
}

func (q *Queue) AddAfter(key string, d time.Duration) {
	q.mu.Lock()
	defer q.mu.Unlock()
	if q.down {
		return
	}
	q.timers = append(q.timers, time.AfterFunc(d, func() { q.Add(key) }))
}

// AddRateLimited requeues with delay base * 2^(failures-1), capped at max.
func (q *Queue) AddRateLimited(key string) time.Duration {
	q.mu.Lock()
	q.failures[key]++
	d := q.base << (q.failures[key] - 1)
	if d > q.max || d <= 0 {
		d = q.max
	}
	q.Backoffs = append(q.Backoffs, d)
	q.mu.Unlock()
	q.AddAfter(key, d)
	return d
}

func (q *Queue) Forget(key string) {
	q.mu.Lock()
	defer q.mu.Unlock()
	delete(q.failures, key)
}

func (q *Queue) NumRequeues(key string) int {
	q.mu.Lock()
	defer q.mu.Unlock()
	return q.failures[key]
}

// Get blocks until a key is available or the queue is shut down.
func (q *Queue) Get() (string, bool) {
	q.mu.Lock()
	defer q.mu.Unlock()
	for len(q.order) == 0 && !q.down {
		q.cond.Wait()
	}
	if len(q.order) == 0 {
		return "", true
	}
	key := q.order[0]
	q.order = q.order[1:]
	delete(q.dirty, key)
	q.processing[key] = true
	return key, false
}

func (q *Queue) Done(key string) {
	q.mu.Lock()
	defer q.mu.Unlock()
	delete(q.processing, key)
	if q.dirty[key] {
		q.order = append(q.order, key)
		q.cond.Signal()
	}
}

func (q *Queue) ShutDown() {
	q.mu.Lock()
	defer q.mu.Unlock()
	q.down = true
	for _, t := range q.timers {
		t.Stop()
	}
	q.cond.Broadcast()
}

// ---------- Event log and fault injection ----------

type Log struct {
	mu      sync.Mutex
	start   time.Time
	entries []string
}

func (l *Log) Printf(format string, a ...any) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.entries = append(l.entries, fmt.Sprintf("%6dms  ", time.Since(l.start).Milliseconds())+fmt.Sprintf(format, a...))
}

func (l *Log) Contains(sub string) bool {
	l.mu.Lock()
	defer l.mu.Unlock()
	for _, e := range l.entries {
		if strings.Contains(e, sub) {
			return true
		}
	}
	return false
}

// Faults holds one-shot errors the scenario injects into Reconcile, keyed by object name.
type Faults struct {
	mu   sync.Mutex
	once map[string]error
}

func (f *Faults) Set(key string, err error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.once[key] = err
}

func (f *Faults) Take(key string) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	err := f.once[key]
	delete(f.once, key)
	return err
}

// ---------- Reconciler: level-based, idempotent, reads from caches, writes with re-read ----------

type Result struct {
	RequeueAfter time.Duration
}

type Reconciler struct {
	apps     *Store[*WebApp]
	deps     *Store[*Deployment]
	appCache *Cache[*WebApp]
	depCache *Cache[*Deployment]
	faults   *Faults
	log      *Log
}

func (r *Reconciler) Reconcile(ctx context.Context, key string) (Result, error) {
	app, ok := r.appCache.Get(key)
	if !ok {
		// Object is gone. Idempotent cleanup in case the finalizer path was skipped.
		if err := r.deps.Delete(key); err == nil {
			r.log.Printf("reconcile %s: webapp gone, deleted orphan deployment", key)
		}
		return Result{}, nil
	}

	if app.DeletionTimestamp != nil {
		switch err := r.deps.Delete(key); {
		case err == nil:
			r.log.Printf("reconcile %s: deleting, owned deployment removed", key)
		case !errors.Is(err, ErrNotFound):
			return Result{}, err
		}
		err := RetryOnConflict(5, func() error {
			latest, err := r.apps.Get(key)
			if err != nil {
				return err
			}
			latest.Finalizers = remove(latest.Finalizers, finalizerName)
			_, err = r.apps.Update(latest)
			return err
		})
		if errors.Is(err, ErrNotFound) {
			r.log.Printf("reconcile %s: already gone; nothing to do (idempotent)", key)
			return Result{}, nil
		}
		if err != nil {
			return Result{}, err
		}
		r.log.Printf("reconcile %s: finalizer removed, object can go", key)
		return Result{}, nil
	}

	if !contains(app.Finalizers, finalizerName) {
		err := RetryOnConflict(5, func() error {
			latest, err := r.apps.Get(key)
			if err != nil {
				return err
			}
			latest.Finalizers = append(latest.Finalizers, finalizerName)
			_, err = r.apps.Update(latest)
			return err
		})
		if err != nil {
			return Result{}, err
		}
		r.log.Printf("reconcile %s: finalizer added", key)
		return Result{}, nil // the Update triggers a new event; next pass continues with fresh data
	}

	if err := r.faults.Take(key); err != nil {
		return Result{}, err
	}

	// Child: create if missing, update if different. Compare desired with observed, never with the event.
	dep, ok := r.depCache.Get(key)
	if !ok {
		created, err := r.deps.Create(&Deployment{Meta: Meta{Name: key}, Owner: key, Replicas: app.Replicas, Image: app.Image})
		if errors.Is(err, ErrAlreadyExists) {
			// Cache lagged behind our own earlier write; read the truth and continue.
			if created, err = r.deps.Get(key); err != nil {
				return Result{}, err
			}
		} else if err != nil {
			return Result{}, err
		} else {
			r.log.Printf("reconcile %s: created deployment replicas=%d image=%s", key, created.Replicas, created.Image)
		}
		dep = created
	}
	if dep.Replicas != app.Replicas || dep.Image != app.Image {
		err := RetryOnConflict(5, func() error {
			latest, err := r.deps.Get(key)
			if err != nil {
				return err
			}
			latest.Replicas, latest.Image = app.Replicas, app.Image
			_, err = r.deps.Update(latest)
			return err
		})
		if err != nil {
			return Result{}, err
		}
		r.log.Printf("reconcile %s: updated deployment to replicas=%d image=%s", key, app.Replicas, app.Image)
	}

	// Status: copy what the child reports. The user may be writing spec at the same time, so re-read.
	if app.ReadyReplicas != dep.ReadyReplicas {
		err := RetryOnConflict(5, func() error {
			latest, err := r.apps.Get(key)
			if err != nil {
				return err
			}
			latest.ReadyReplicas = dep.ReadyReplicas
			if _, err = r.apps.Update(latest); errors.Is(err, ErrConflict) {
				r.log.Printf("reconcile %s: status update hit %v; re-reading and retrying", key, ErrConflict)
			}
			return err
		})
		if err != nil {
			return Result{}, err
		}
		r.log.Printf("reconcile %s: status readyReplicas=%d", key, dep.ReadyReplicas)
	}
	if dep.ReadyReplicas != dep.Replicas {
		return Result{RequeueAfter: 200 * time.Millisecond}, nil // rollout in progress: look again soon
	}
	return Result{}, nil
}

func contains(list []string, s string) bool {
	for _, x := range list {
		if x == s {
			return true
		}
	}
	return false
}

func remove(list []string, s string) []string {
	out := list[:0]
	for _, x := range list {
		if x != s {
			out = append(out, x)
		}
	}
	return out
}

// ---------- Controller: informers feed the queue, workers call Reconcile, clean shutdown ----------

type Controller struct {
	r       *Reconciler
	queue   *Queue
	workers int
}

func (c *Controller) Run(ctx context.Context, wg *sync.WaitGroup) {
	// Watch WebApps by name and Deployments by owner: a child event enqueues its parent.
	wg.Add(2 + c.workers)
	go func() {
		defer wg.Done()
		RunInformer(ctx, c.r.apps, c.r.appCache, func(a *WebApp) { c.queue.Add(a.Name) })
	}()
	go func() {
		defer wg.Done()
		RunInformer(ctx, c.r.deps, c.r.depCache, func(d *Deployment) { c.queue.Add(d.Owner) })
	}()
	for i := 0; i < c.workers; i++ {
		go func() {
			defer wg.Done()
			c.worker(ctx)
		}()
	}
	wg.Add(1)
	go func() {
		defer wg.Done()
		<-ctx.Done()
		c.queue.ShutDown() // wakes every worker blocked in Get
	}()
}

func (c *Controller) worker(ctx context.Context) {
	for {
		key, down := c.queue.Get()
		if down {
			return
		}
		res, err := c.r.Reconcile(ctx, key)
		switch {
		case err != nil:
			d := c.queue.AddRateLimited(key)
			c.r.log.Printf("reconcile %s failed: %v; requeue in %v (failure %d)", key, err, d, c.queue.NumRequeues(key))
		case res.RequeueAfter > 0:
			c.queue.Forget(key)
			c.queue.AddAfter(key, res.RequeueAfter)
		default:
			c.queue.Forget(key)
		}
		c.queue.Done(key)
	}
}

// ---------- Fake kubelet: moves ReadyReplicas toward Replicas, one pod per tick ----------

func runKubelet(ctx context.Context, deps *Store[*Deployment], log *Log) {
	tick := time.NewTicker(30 * time.Millisecond)
	defer tick.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-tick.C:
			deps.mu.Lock()
			names := make([]string, 0, len(deps.items))
			for n := range deps.items {
				names = append(names, n)
			}
			deps.mu.Unlock()
			for _, n := range names {
				_ = RetryOnConflict(5, func() error {
					d, err := deps.Get(n)
					if err != nil || d.ReadyReplicas == d.Replicas {
						return nil
					}
					if d.ReadyReplicas < d.Replicas {
						d.ReadyReplicas++
					} else {
						d.ReadyReplicas = d.Replicas
					}
					if _, err = deps.Update(d); err == nil {
						log.Printf("kubelet: deployment %s readyReplicas=%d/%d", n, d.ReadyReplicas, d.Replicas)
					}
					return err
				})
			}
		}
	}
}

// ---------- Scenario ----------

var failures []string

func check(ok bool, what string) {
	if ok {
		fmt.Println("check ok:", what)
	} else {
		failures = append(failures, what)
		fmt.Println("CHECK FAILED:", what)
	}
}

func waitFor(what string, cond func() bool) bool {
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		if cond() {
			return true
		}
		time.Sleep(5 * time.Millisecond)
	}
	fmt.Println("timeout waiting for:", what)
	return false
}

func main() {
	before := runtime.NumGoroutine()
	log := &Log{start: time.Now()}
	apps, deps := NewStore[*WebApp]("webapp"), NewStore[*Deployment]("deployment")
	r := &Reconciler{apps: apps, deps: deps, appCache: NewCache[*WebApp](), depCache: NewCache[*Deployment](),
		faults: &Faults{once: map[string]error{}}, log: log}
	ctrl := &Controller{r: r, queue: NewQueue(5*time.Millisecond, time.Second), workers: 2}

	ctx, cancel := context.WithCancel(context.Background())
	var wg sync.WaitGroup
	ctrl.Run(ctx, &wg)
	wg.Add(1)
	go func() {
		defer wg.Done()
		runKubelet(ctx, deps, log)
	}()

	// Stage 1: create a WebApp with 2 replicas; expect a Deployment, a finalizer and status 2/2.
	_, err := apps.Create(&WebApp{Meta: Meta{Name: "shop"}, Replicas: 2, Image: "shop:v1"})
	check(err == nil, "stage 1: webapp shop created")
	ready := func(n int, image string) bool {
		a, err1 := apps.Get("shop")
		d, err2 := deps.Get("shop")
		return err1 == nil && err2 == nil && d.Owner == "shop" && d.Replicas == n && d.Image == image &&
			d.ReadyReplicas == n && a.ReadyReplicas == n
	}
	check(waitFor("stage 1", func() bool { return ready(2, "shop:v1") }), "stage 1: deployment shop has 2/2 ready and webapp status says 2")
	a, _ := apps.Get("shop")
	check(contains(a.Finalizers, finalizerName), "stage 1: finalizer "+finalizerName+" is set")

	// Stage 2: scale to 3. Then make another client touch the WebApp right before the controller's
	// status write, so that write gets a Conflict and must re-read.
	err = RetryOnConflict(5, func() error {
		latest, err := apps.Get("shop")
		if err != nil {
			return err
		}
		latest.Replicas = 3
		_, err = apps.Update(latest)
		return err
	})
	check(err == nil, "stage 2: user scaled webapp to 3 replicas")
	var injected atomic.Int32
	apps.SetBeforeWrite(func() {
		latest, _ := apps.Get("shop")
		latest.Annotations = map[string]string{"touched-by": "another-client"}
		if _, err := apps.Update(latest); err == nil {
			injected.Add(1)
			log.Printf("other client: wrote annotation on webapp shop (rv now %d)", latest.ResourceVersion+1)
		}
	})
	check(waitFor("stage 2", func() bool { return ready(3, "shop:v1") }), "stage 2: deployment scaled to 3/3 and status says 3")
	appConflicts, _ := apps.Stats()
	check(injected.Load() == 1 && appConflicts >= 1, fmt.Sprintf("stage 2: injected 1 concurrent write, store rejected %d update(s) with Conflict", appConflicts))
	check(log.Contains("re-reading and retrying"), "stage 2: controller re-read after the Conflict and the retry succeeded")

	// Stage 3: a transient error on the next reconcile; expect a rate-limited requeue, then success.
	r.faults.Set("shop", errors.New("simulated API timeout"))
	err = RetryOnConflict(5, func() error {
		latest, err := apps.Get("shop")
		if err != nil {
			return err
		}
		latest.Image = "shop:v2"
		_, err = apps.Update(latest)
		return err
	})
	check(err == nil, "stage 3: user changed image to shop:v2")
	check(waitFor("stage 3", func() bool { return ready(3, "shop:v2") }), "stage 3: deployment rolled to shop:v2 after the transient error")
	check(len(ctrl.queue.Backoffs) == 1 && ctrl.queue.Backoffs[0] == 5*time.Millisecond,
		fmt.Sprintf("stage 3: exactly one backoff requeue, delay %v", ctrl.queue.Backoffs))
	check(ctrl.queue.NumRequeues("shop") == 0, "stage 3: failure count forgotten after the successful reconcile")

	// Stage 4: delete. The finalizer holds the object until the Deployment is gone.
	check(apps.Delete("shop") == nil, "stage 4: delete requested")
	check(waitFor("stage 4", func() bool {
		_, e1 := apps.Get("shop")
		_, e2 := deps.Get("shop")
		return errors.Is(e1, ErrNotFound) && errors.Is(e2, ErrNotFound)
	}), "stage 4: deployment removed first, then finalizer cleared and webapp gone")
	check(log.Contains("finalizer removed"), "stage 4: event log shows the finalizer being removed")

	// Stage 5: shutdown. Cancel the context, wait for every goroutine, compare goroutine counts.
	running := runtime.NumGoroutine()
	cancel()
	wg.Wait()
	time.Sleep(20 * time.Millisecond) // let any timer callback that fired during shutdown return
	after := runtime.NumGoroutine()
	appConflicts, appDropped := apps.Stats()
	depConflicts, depDropped := deps.Stats()
	check(appDropped == 0 && depDropped == 0, "no watch events were dropped")
	check(after == before, fmt.Sprintf("goroutines: %d before start, %d while running, %d after shutdown", before, running, after))

	fmt.Println("\nevent log:")
	for _, e := range log.entries {
		fmt.Println(" ", e)
	}
	fmt.Printf("\nstore conflicts: webapp=%d deployment=%d (the deployment ones are natural races with the kubelet)\n", appConflicts, depConflicts)
	if len(failures) > 0 {
		fmt.Printf("%d check(s) failed\n", len(failures))
		os.Exit(1)
	}
	fmt.Println("all checks passed")
}