Desired-state reconciliation

Deep dive

How it works

Parts:

• spec store — desired state per key with resource_version and generation (bumped only when spec changes)
• watch cache — local copy fed by watch events plus periodic full resync
• work queue — deduped keys, per-key exponential backoff, global rate limit
• reconciler — read spec and actual, diff, act idempotently, write status only if changed
• owner references and finalizers — dependents are garbage-collected with the owner; a finalizer holds deletion until external cleanup is done
• leader lease — one active replica per controller

1. A watch event or resync tick enqueues the object's key, not the event.
2. A worker pops a key; duplicates of waiting keys were dropped.
3. Reconcile reads the latest spec and actual state, whatever woke it.
4. It creates or deletes only the difference; a second run changes nothing.
5. It writes status (observedGeneration, counts) only if it differs.
6. On error the key returns after 2^n ticks (capped); success resets n.
7. Writes carry the resource_version read; a mismatch is a conflict: re-read, retry.
8. Delete sets a deletion timestamp; the controller cleans up external resources, removes its finalizer, then the object goes.

 user/CI ──put spec(rv)──► [spec store] ──watch──► [cache] ──key──► [queue: dedupe+backoff]
                                ▲                      ▲  resync          │
             status(observedGen)│                      │                  ▼
                           [reconciler] ◄── read ── [actual world] ◄── create/delete

Code

Toy: 8 apps, pods die, 30 % of events are lost, 10 % of reconciles fail. Level-triggered (queue + resync) converges; edge-triggered (applies each event's delta) does not.

use std::collections::{HashMap, HashSet, VecDeque};
struct Rng(u64);
impl Rng {
    fn pct(&mut self, p: u64) -> bool {
        self.0 ^= self.0 << 13; self.0 ^= self.0 >> 7; self.0 ^= self.0 << 17;
        self.0 % 100 < p
    }
    fn below(&mut self, n: usize) -> usize { self.pct(0); (self.0 % n as u64) as usize }
}
#[derive(Clone, Debug)]
struct Spec { replicas: u32, generation: u64, rv: u64 }
#[derive(Clone, PartialEq, Debug)]
struct Status { observed_generation: u64, ready: u32 }
#[derive(Debug, PartialEq)] enum StoreErr { Conflict }
#[derive(Default)]
struct Store { specs: HashMap<String, Spec>, status: HashMap<String, Status>, rv: u64, writes: u64 }
impl Store {
    /// Optimistic concurrency: the write must name the version it read.
    fn put_spec(&mut self, k: &str, replicas: u32, expect: Option<u64>) -> Result<u64, StoreErr> {
        let cur = self.specs.get(k).cloned();
        if let (Some(c), Some(e)) = (&cur, expect) { if c.rv != e { return Err(StoreErr::Conflict); } }
        self.rv += 1; self.writes += 1;
        let generation = cur.map_or(1, |c| c.generation + (c.replicas != replicas) as u64);
        self.specs.insert(k.into(), Spec { replicas, generation, rv: self.rv });
        Ok(self.rv)
    }
    fn set_status(&mut self, k: &str, s: Status, skip_noop: bool) {
        if skip_noop && self.status.get(k) == Some(&s) { return; }
        self.rv += 1; self.writes += 1;
        self.status.insert(k.into(), s);
    }
}
/// Work queue: per-key dedupe, exponential backoff in ticks (cap 32).
struct Queue { dedupe: bool, ready: VecDeque<String>, queued: HashSet<String>, fails: HashMap<String, u32>, later: Vec<(u64, String)> }
impl Queue {
    fn add(&mut self, k: &str) { if !self.dedupe || self.queued.insert(k.into()) { self.ready.push_back(k.into()); } }
    fn get(&mut self, now: u64) -> Option<String> {
        let due: Vec<String> = self.later.iter().filter(|(t, _)| *t <= now).map(|(_, k)| k.clone()).collect();
        self.later.retain(|(t, _)| *t > now);
        for k in due { self.add(&k); }
        let k = self.ready.pop_front()?;
        self.queued.remove(&k);
        Some(k)
    }
    fn fail(&mut self, k: &str, now: u64) {
        let n = self.fails.entry(k.into()).or_insert(0);
        *n += 1;
        self.later.push((now + (1u64 << (*n - 1).min(5)), k.into()));
    }
}
struct Cfg { level: bool, dedupe: bool, skip_noop: bool, resync: u64 }
struct Out { converged: bool, reconciles: u64, writes: u64 }
/// Idempotent: derive the diff from full state, act, record what was observed.
fn reconcile(st: &mut Store, world: &mut HashMap<String, u32>, k: &str, skip: bool) {
    let want = st.specs[k].replicas;
    let have = world.entry(k.into()).or_insert(0);
    *have = want; // stands for creating/deleting the missing pods
    let g = st.specs[k].generation;
    st.set_status(k, Status { observed_generation: g, ready: want }, skip);
}
fn simulate(c: &Cfg, seed: u64) -> Out {
    let (mut rng, mut st) = (Rng(seed), Store::default());
    let apps: Vec<String> = (0..8).map(|i| format!("app{i}")).collect();
    let mut world: HashMap<String, u32> = HashMap::new();
    for a in &apps { st.put_spec(a, 3, None).unwrap(); world.insert(a.clone(), 3); }
    let mut q = Queue { dedupe: c.dedupe, ready: VecDeque::new(), queued: HashSet::new(), fails: HashMap::new(), later: vec![] };
    let mut reconciles = 0;
    for t in 0..300u64 {
        let mut events: Vec<(String, i32)> = vec![];
        if t < 200 {
            for _ in 0..3 {
                let a = apps[rng.below(8)].clone();
                if world[&a] > 0 { *world.get_mut(&a).unwrap() -= 1; events.push((a, 1)); }
            }
            if t % 25 == 0 {
                let a = apps[rng.below(8)].clone();
                let (old, rv) = (st.specs[&a].replicas, st.specs[&a].rv);
                let new = 1 + rng.below(5) as u32;
                st.put_spec(&a, new, Some(rv)).unwrap();
                events.push((a, new as i32 - old as i32));
            }
        }
        for (a, delta) in events {
            if rng.pct(30) { continue; }
            if c.level { q.add(&a); }
            else { let w = world.get_mut(&a).unwrap(); *w = (*w as i32 + delta).max(0) as u32; }
        }
        if c.level {
            if t % c.resync == 0 { for a in &apps { q.add(a); } }
            for _ in 0..4 {
                let Some(k) = q.get(t) else { break };
                reconciles += 1;
                if rng.pct(10) { q.fail(&k, t); } else { q.fails.remove(&k); reconcile(&mut st, &mut world, &k, c.skip_noop); }
            }
        }
    }
    let converged = apps.iter().all(|a| world[a] == st.specs[a].replicas);
    Out { converged, reconciles, writes: st.writes }
}
fn main() {
    let lvl = Cfg { level: true, dedupe: true, skip_noop: true, resync: 20 };
    let edge = Cfg { level: false, ..lvl };
    for seed in 1..=20u64 { assert!(simulate(&lvl, seed * 7919).converged, "level seed {seed}"); }
    let drifted = (1..=20u64).filter(|s| !simulate(&edge, s * 7919).converged).count();
    assert!(drifted >= 15, "edge-triggered drifted in {drifted}/20");
    let mut st = Store::default(); let mut w = HashMap::new();
    st.put_spec("a", 3, None).unwrap();
    reconcile(&mut st, &mut w, "a", true); let after = st.writes;
    reconcile(&mut st, &mut w, "a", true);
    assert_eq!(st.writes, after);
    let rv = st.specs["a"].rv; st.put_spec("a", 5, Some(rv)).unwrap();
    assert_eq!(st.put_spec("a", 7, Some(rv)), Err(StoreErr::Conflict)); // stale version rejected
    let (base, no_dedupe, no_skip) = (simulate(&lvl, 42), simulate(&Cfg { dedupe: false, ..lvl }, 42), simulate(&Cfg { skip_noop: false, ..lvl }, 42));
    println!("edge drifted {drifted}/20; reconciles {} / {} (no dedupe); writes {} / {} (no noop-skip)", base.reconciles, no_dedupe.reconciles, base.writes, no_skip.writes);
    assert!(no_dedupe.reconciles > base.reconciles && no_skip.writes > base.writes);
}

Complexity and constants

One reconcile is O(children of the key), independent of how many events woke it. Cost drivers: API round trips (cache read = memory; live GET/list = network), writes (each a store commit, usually a quorum fsync) and empty wake-ups. Toy (8 apps, 300 ticks, seed 42): 499 reconciles, 28 writes; 588 reconciles without dedupe; 463 writes without the status no-op check.

Efficiency

• Cost: one reconcile = cache reads (order of µs) plus the writes that change something; the bound is store write latency, not CPU. A converged key costs zero writes.
• Levers:
  • informer cache instead of GET/poll — reads from memory, one list+watch per type at the API server (client-go) — always.
  • per-key dedupe — N events, one pass (toy: 588 → 499 reconciles, measured) — bursty events.
  • skip unchanged status writes — toy: 463 → 28 writes (measured) — status has no per-pass timestamps.
  • resync period — controller-runtime default 10 h with 10 % jitter (cache docs); shorter repairs lost events sooner but re-reconciles everything.
  • backoff and rate limit — client-go default: per-key exponential 5 ms to 1000 s, plus a 10 qps / burst 100 bucket (source) — failing dependencies.
• Naive → standard → state of the art:
  • poll all objects each second, rewrite status · O(objects) reads and writes per second.
  • watch cache + deduped queue + jittered resync · memory reads, writes only on change.
  • key-sharded controllers, lease-elected standbys, patch updates · parallel, few conflicts.

Implementation notes

• Queue by namespace/name only; fetch fresh state on pop.
• A key re-added while being processed must run again afterwards (client-go: "dirty" flag; the toy drops it from the set on pop).
• Compare spec fields, never put changing timestamps in status, or each write retriggers the watch (hot loop).
• Wait for the cache to sync before the first reconcile.
• Fence side effects that outlive a lost lease (fencing tokens).

Pitfalls

• Acting on the event ("+1") instead of the diff reintroduces drift.
• Status written from a stale read: require the version (conditional writes).
• Random child names duplicate resources on retry.
• A finalizer nobody removes leaves the object Terminating forever.

Variants and alternatives

In production

• Kubernetes controllers and operators (Burns et al.), including database operators such as CloudNativePG.
• Argo CD and Flux reconcile clusters from Git; Crossplane, cloud resources.
• AWS IoT Device Shadow: desired and reported device state with versions.

Implementations

Sources

• Burns, Grant, Oppenheimer, Brewer, Wilkes, Borg, Omega, and Kubernetes, ACM Queue 14(1), 2016. page
• Kubernetes docs: Controllers · Finalizers · Garbage collection · API concepts · Leases
• client-go workqueue · Kubebuilder book
• Terraform drift · Argo CD · Flux · AWS IoT Device Shadow