Atomicity vs visibility — the invariant's shape picks volatile, an atomic, or a lock

The idea

The invariant's shape picks the primitive

Both deliberately broken variants of the corpus rate limiter pass its base suite 28 of 28. That is why this lesson exists. Atomicity bugs and visibility bugs are invisible to every test you have time to write in a round, and they are different problems with different fixes.

The move that costs candidates the concurrency leg is volatile on a counter. It fixes the wrong problem. Measured in worked/: a ceiling of 20000, two hundred threads, volatile int with check-then-act — 20028 to 21553 admitted across fourteen runs, never the limit. Every read was fresh. The read and the write were still two operations.

The decision takes ten seconds. Name each piece of shared mutable state, then match its shape. A flag one thread writes and others only read: volatile, and that is its whole job. One field whose new value depends on its old one: an Atomic*, in one call — getAndUpdate, merge, computeIfAbsent. A rule that spans two fields, or a check in one call and a write in another: a lock, held around every read and write that participates.

Each arm is proven in worked/ with the failure it prevents. contrast/ then measures both overshoots. The all-volatile design passes the base suite and admits up to 906 extra under storm. A global lock is exactly correct and fourteen times slower than one lock per budget.

J12 mapped these primitives from C++. This lesson is the choosing, and the invariant decides — never the field.

Worked walkthrough

NOTES — four decisions, and the measurement behind each one

Run it first

..\..\..\..\.toolchain\jdk-21\bin\javac.exe -Xlint:all -d out *.java
..\..\..\..\.toolchain\jdk-21\bin\java.exe -cp out Main

javac -Xlint:all prints nothing and exits 0. That is the first fact worth sitting with: every broken class in this lesson compiles clean, and there is no flag that finds any of it.

Real output, verbatim, from one run:

crowd = 200 threads

--- 1. a ceiling of 20000, 200 threads, 200 attempts each
  volatile int, check then act   admitted 20060  (over by 60)
  AtomicInteger.getAndUpdate     admitted 20000  (over by 0)

--- 2. burst 200 + hourly 300 per client, all-or-nothing, 50 clients, 200 threads x 2 attempts each
  plain ints, no lock          admitted 10202 of 10000 allowed (over by 202); hourly rules charged 10028 for 10202 admitted; torn snapshots seen: 42957; unbuildable Verdicts: 0
  one AtomicInteger per field  admitted 10000 of 10000 allowed (over by 0); torn snapshots seen: 2036; boundary hammer breached 1 of 30 single-budget storms
  one lock over the decision   admitted 10000 of 10000 allowed (over by 0); torn snapshots seen: 0; boundary hammer breached 0 of 30 single-budget storms

--- 3. registering a counter on a ConcurrentHashMap: 20 fresh keys, 200 threads x 50 records each
  get, null-check, put            counted 199092 of 200000  (lost 908)
  computeIfAbsent                 counted 200000 of 200000  (lost 0)

--- 4. close() sets a stop flag and waits up to 1500ms for the worker
  plain boolean      worker exited: false
  volatile boolean   worker exited: true  (served 312630243 polls before stopping)

Which numbers you can rely on, measured over fourteen runs

A demonstration of a race that only sometimes races teaches the wrong thing, so here is the spread rather than one lucky figure.

The two rows that say no matter more than the rows that say yes. The per-field-atomics budget passed its end-state check on ten runs of fourteen. A candidate who tests it casually concludes it works; a graded suite that storms it enough times concludes otherwise, and both are looking at the same class. Rarity is camouflage, not safety. The observer row directly above it is the reason the design is wrong regardless of what the end state says.

The rule the four blocks add up to

Name each piece of shared mutable state, then ask three questions about it. A field one thread writes and others only read, with no arithmetic on the old value: volatile. One field whose new value depends on its old value: an Atomic*, in one call. A rule that spans more than one field, or a check in one call and a write in another: a lock, held around every read and write that participates. J12 covers what each primitive is; this lesson is about choosing, and the choosing is decided by the shape of the invariant, never by the type of the field.

BurstCeiling.java

VolatileGuess

private volatile int used;

if (used < limit) {
    used++;
    return true;
}

What volatile buys here: every thread's read of used is fresh. No stale values, no hoisting, full ordering. That is visibility, and visibility was not the problem.

What it cannot buy: the if and the ++ are still separate. Forty threads can pass the check before any of them writes, and each is then entitled to admit. used++ is itself a read, an add and a store, so updates are lost on top of the over-admission. Block 1 measured 28 to 1553 extra admissions across fourteen runs, never zero. A candidate who writes volatile on a counter has changed which wrong number appears, and nothing else.

OneAtomic

return used.getAndUpdate(u -> u < limit ? u + 1 : u) < limit;

The whole decision in one operation, which is what makes it a guarantee. getAndUpdate retries a compare-and-set until the function applies to an unchanged value, then returns what it replaced. Admitted exactly when the replaced value was under the limit. This is the compare_exchange_weak retry loop you know from C++, with the loop already written.

used can never exceed limit, and that is load-bearing for readers. A version built on incrementAndGet() <= limit admits the same requests and leaves used at 40000 after the storm, so any remaining()-style read reports a debt the client does not owe. The faded stage's GapTest checks this with a plain sequential over-subscription, no threads needed.

The threshold for this arm: one field, one rule mentioning only that field. The moment a second field joins the rule, no atomic can follow it there. That is block 2.

DualGate.java

Verdict

public record Verdict(boolean allowed, int remainingHour) {
    public Verdict {
        if (remainingHour < 0) { throw new IllegalArgumentException(...); }
    }
}

The invariant lives in the value type, so a torn count cannot travel. This is A5 paying rent under concurrency. corpus/rate-limiter's tuning log records its Decision throwing remaining cannot be negative, got -3 during a stress run of a broken variant — an invariant catching a concurrency bug no test was aimed at. Main counts these; in the shipped configuration no storm drove an hourly counter past its own limit, so the count printed 0 on all fourteen runs. The check still earns its line: it turns a silent lie into a loud failure whenever a race lands there, and it costs nothing on the paths where none does.

NoLock

if (burstUsed < burstLimit && hourUsed < hourLimit) {
    burstUsed++;
    hourUsed++;

Nothing joins the check to the charge, and nothing joins the two charges to each other. Three failures, all measured in block 2. It over-admits: 13 to 202 past the ceiling. Its two counters drift apart, because each ++ loses updates independently: the hourly rules ended 11 to 174 charges short of what was admitted. And an observer reading the pair mid-storm saw it half-charged tens of thousands of times.

This class passes any test that calls it from one thread. Sequentially it is exactly correct, which is why contrast/'s base suite is green against a design with this bug. The corpus's own broken rate-limiter variants pass their base suite 28 of 28 for the same reason.

PerFieldAtomics

private final AtomicInteger burstUsed = new AtomicInteger();
private final AtomicInteger hourUsed = new AtomicInteger();

Every individual operation is now atomic, and the decision is still four of them. The check reads two counters; the charge writes two; other threads land between any pair. The honest measurement is in the table above: the end state is usually exact and sometimes is not, which is the worst possible testing profile.

The reliable evidence against it is the observer, not the end state. All-or-nothing means "burst charged" and "hour charged" are equal at every observable moment. The observer caught them unequal 159 to 3707 times per storm, every run. There is no way to update the pair as one moment, and none to read it as one moment either. charges() here is two get()s with a gap, the same bug wearing a reader's clothes.

An atomic guards a field. A lock guards a rule. That sentence is the answer to the interviewer's "why not an AtomicLong?". corpus/rate-limiter/reference/DECISION_LOG.md states the full version: the unit that must be atomic is check-every-rule-then-charge-every-rule, and no concurrent collection or atomic can express that.

Locked

private final Object lock = new Object();
private int burstUsed;
private int hourUsed;

One lock, because the two fields move under one rule. private, so no caller can name the monitor and take it — J12's block 6 shows a stranger doing exactly that to a synchronized method. final, so two threads can never lock different objects while both believe they guard these fields.

Plain fields inside, and the omission is the point. Releasing a monitor happens-before the next acquisition of the same monitor, so every write is visible to the next thread in. volatile under a lock is not extra safety; it is a signal you did not know which mechanism was working.

charges() takes the lock for a read. A read that reports on locked state is a participant in the invariant. corpus/lru-cache makes get synchronized for the same reason, and its DECISION_LOG records what happened to the variant that skipped it: expected 20, was 119.

ClientCounts.java

Racy

AtomicLong counter = counts.get(client);
if (counter == null) {
    counter = new AtomicLong();
    counts.put(client, counter);
}
counter.incrementAndGet();

Both map calls are atomic; the null check between them is not. Two hundred threads hit a fresh key, all see null, all put. Last write wins, and every increment made against a replaced counter vanishes with it. Block 3 lost 268 to 1337 increments per run, every run.

The increments themselves were never the problem — AtomicLong did its job. What raced was deciding which object to increment. Check-then-act survives review because the field's type says Concurrent and the reviewer's eye stops there. It is the highest-frequency real bug in graded concurrency legs.

OneCall

counts.computeIfAbsent(client, key -> new AtomicLong()).incrementAndGet();

One call, so there is no gap for a second registration to land in. The function runs while the map holds that bin's lock; exactly one counter can ever exist per key. 200000 of 200000 on every run. The family to reach for whenever an update reads the map first: merge, compute, computeIfAbsent, computeIfPresent, putIfAbsent, replace.

The cost, so the choice stays a choice. The function runs under a bin lock, so it must be short, must not block, and must not touch the same map. This exact shape, computeIfAbsent to an AtomicLong followed by one atomic increment, is what corpus/rate-limiter/curveballs/03-denials-are-counted ships as its reference patch.

Drain.java

The failure here has no lost update in it, which is why it needs its own block. Nothing is miscounted; the numbers are all fine; the process cannot shut down.

private boolean stopped;

while (!stopped) { n++; }

Nothing in the worker writes stopped, so the JIT may read it once and keep it in a register. That is a legal transformation for a plain field. The loop becomes infinite the moment it is compiled, and close() then writes a flag no one will read again. Fourteen runs: the plain worker outlived its 1500ms grace on every one. The worker is a daemon thread only so the demo can exit; a real appender's worker would hang the JVM at shutdown, with nothing on any stack to say why.

private volatile boolean stopped;

One keyword, and the read cannot be hoisted or go stale. Fourteen runs, exited every time, after a few hundred million polls. This is the shape volatile exists for, and all three of its conditions hold: one writer, a write that ignores the current value, no sibling field that must change with it. Miss any one and you are back in blocks 1 and 2.

served is a plain long, and leaving it plain is correct. It is written by one thread and read only after close() observes the join. Thread.join is a happens-before edge, so the read needs nothing — protecting state that is never shared while two threads run is cost without a customer, which is when-not.md's subject.

Main.java

ready.countDown();
go.await();

Three latches, so the threads collide instead of drifting past each other. No thread starts until all 200 stand at the line. This is the same design as harness/DojoConcurrency.java, which the graded suites use.

int mine = 0;
...
admitted.addAndGet(mine);

Each thread tallies locally and publishes once, and this is a measurement lesson learned the hard way. The first draft bumped a shared AtomicInteger per admission, inside the loop. That serialised the threads on the instrument, and the no-lock budget breached in only 2 of 40 storms. Counting locally took the instrument out of the timing, and the breach became a 14-of-14-runs fact. An observer that synchronises the observed is not observing it.

int keys = 50;

Fifty budgets, because a race needs chances. One budget saturates in the first moments of a storm, while most of the crowd is still waking, so its boundary is crossed by a handful of hot threads once. Fifty budgets is fifty boundary crossings under full contention. The corpus tuning logs tell the same story as catch rates. lru-cache's unsynchronized get was caught 3 of 10 times at CROWD=4 and 9 of 10 at CROWD=200. The shipped suites repeat every scenario, because one quiet run proves nothing.

When not to

When not to synchronize

Every mechanism in this lesson has a cost, and the costs are why "sprinkle synchronized until the suite goes green" loses marks even when it works. STANDARD v1.0 scores the concurrency answer by whether you can say what a lock protects. Protection with no invariant behind it fails that question from the other side.

The concrete bad example, measured

contrast/b-coarse/Throttle.java is the fix applied too widely. It compiles clean, and CurveballTest passes it on every run. The change from b/ is eight lines:

private static final Object LOCK = new Object();

One monitor for every client budget in the process, instead of one per budget. What it buys: nothing — b/ was already exact on every storm, so there is no correctness left to purchase. What it costs, from node lessons/E2/contrast/measure.mjs, eight clients on eight budgets that share no state at all:

b/        median 13 ms
b-coarse/ median 185 ms

Fourteen times slower on this machine, because acme's requests now queue behind globex's. The limiter becomes the bottleneck it was installed to prevent, which is the exact sentence corpus/rate-limiter/reference/DECISION_LOG.md uses to justify one lock per key. Lock width is a design decision with a measurable price. The invariant, two counters inside one budget, names the correct width for you.

State that needs nothing, and what guarding it signals

Confined state. Drain.served in worked/ is written by one thread and read only after close() observes the join. Thread.join is a happens-before edge, so a plain long is correct. The same edge is why Main reads its tallies after the pool drains with no lock, and why both corpus stress suites read their results through join rather than through locks. Wrapping such a field in synchronized or an AtomicLong is not caution. It tells a reviewer you cannot tell shared state from private state, the E1 skill this lesson builds on.

Fields already under a lock. DualGate.Locked keeps plain ints, and J12 says why in one line. volatile on a field the lock already guards is noise, and it suggests you were not sure which mechanism was doing the work.

Immutable state. burstLimit and hourLimit are final and need no protection from any number of threads. A correctly constructed object's final fields are safely published; guarding reads of them is pure cost.

Single-threaded contracts. Several corpus problems never promise concurrent callers, and their references contain no locks at all. Synchronization added there defends against threads that do not exist, and it is paid for in clock — D2 is 25% of the score, and interviewers run main first. Say "this is single-threaded by contract; here is what I would guard if that changed" and spend the minutes on the driver.

The threshold, from both sides

Protect exactly the state one invariant spans, in the narrowest mechanism that joins the whole invariant. That means volatile for the lone flag, one atomic call for the lone counter, and one lock per invariant for anything wider. Then write the sentence that names the state and its guard. corpus/lru-cache's javadoc is the model: "they change together, so they share exactly one lock". Producing that sentence is what the design review is actually testing.

Do not protect what no invariant touches: confined fields, final fields, values read only after a join, or independent budgets that happen to share a class. If you cannot finish the sentence "this lock protects …" with named state and the rule that ties it together, the lock is not protecting anything. Delete it, or find the invariant it was supposed to serve.

The contrast pair

The measured trio: one budget class, three protection choices

Three versions of the same Throttle, a client budget with a burst rule and an hourly rule and an all-or-nothing contract. a/ marks both counters volatile and checks before charging. b/ puts one private lock around the whole decision. b-coarse/ is b/ with the lock made static — one monitor for every client in the process. All three compile clean, and all three pass BaseTest.java. The comparison is between designs a reviewer would plausibly wave through.

The curveball, in the interviewer's words

Looks right single-threaded. Now, this sits in front of our API gateway: two hundred worker threads call tryAcquire on the same throttles at the same time. Limits are hard ceilings. If a client's burst rule says 1600, the 1601st concurrent request is refused. A request the burst rule refuses must not have touched the hourly quota. Convince me.

That is not a new requirement. It is the deployment the old requirement always implied, which is what makes E2 different from every seam lesson: the curveball arrives without a single line of the spec changing.

The numbers

Run it yourself — one command reproduces everything below:

node lessons/E2/contrast/measure.mjs

The race, observed. Probe.java is CurveballTest's storm as a main: 50 throttles, 200 threads, 16 attempts per thread per throttle, limits placed mid-storm. Against a/, one probe run printed:

storm 1: admitted 83307 of 80000 allowed (over by 3307), burst counters 80029,
         hour counters 81949, clients whose two rules disagree: 45
storm 2: admitted 80000 of 80000 allowed (over by 0), ...disagree: 0
storm 3: admitted 82344 of 80000 allowed (over by 2344), ...disagree: 50

Against b/: exactly 80000, both counters exactly 80000, zero disagreements, every storm.

The breach is intermittent, and that is a finding, not a flaw. Across 18 probed storms of a/ while authoring, 10 breached, with the excess ranging from 2 to 3307; whole batches of storms came out clean. That intermittency is why CurveballTest runs twelve rounds — the same reason every corpus stress suite repeats its scenarios. It caught a/ on all fourteen authoring runs, on the first round every time, and passed b/ and b-coarse/ on every run.

The fix, priced by the D4 instrument. measureChange from server/lib/diff.mjs:

The wrong way, priced in time. The 8-line change to b-coarse/ breaks nothing, and CurveballTest stays green against it. Bench.java measures what it costs: eight clients, each with its own budget and its own thread, no shared state between them at all.

b/        rep times 12-20 ms, median 13 ms
b-coarse/ rep times 180-388 ms, median 185 ms

Fourteen times slower on this machine, medians of five repetitions per tree. Wall-clock numbers move with the machine; the ratio is the finding. The reason is structural, not incidental: under b-coarse/, acme's requests wait for globex's, and the limiter becomes the queue it was installed to prevent. corpus/rate-limiter's decision log makes the same call before any benchmark: one lock per key, never per process. Its reason is the sentence "two keys have nothing in common and must not queue behind each other."

What each design gets wrong

a/ is not a careless design. It names the shared fields, reaches for a concurrency keyword, and passes every sequential test. Its two mistakes are the two halves of this lesson. volatile gives each read freshness and gives the read-modify-write nothing, so each ++ loses updates. And the decision spans two fields under one rule, so even perfect per-field atomicity cannot make check-both-then-charge-both one moment. Swap the volatiles for AtomicIntegers and the worked example's Main measures exactly that.

b-coarse/ is not a wrong answer to the correctness question. It is a wrong answer to a question nobody asked. Its lock guards state that never spans instances, so the extra width buys zero additional correctness; both trees are exact on every storm. The price is a fourteen-fold slowdown from making strangers queue. Synchronization has one job: hold still exactly the state one invariant spans. Wider than the invariant is not safer than the invariant.

b/ is the shape to remember. One sentence justifies it in review: these two counters change together under one rule, so they share exactly one lock, and each budget brings its own. That is KeyBudget in corpus/rate-limiter/reference/src/, at lesson scale.

What this pair does not show

The base suite passing a/ is the argument for stress suites, not against base suites — they check different promises. And nothing here needed ReentrantLock, ReadWriteLock, or a concurrent collection. When the invariant is two ints wide, synchronized on a private final Object is the whole answer, and reaching past it is when-not.md's subject.

Worked source

The 5 files of the worked design

Every file below is the one the app opens, verbatim. This is the part worth reading slowly: the prose above argues for a shape, and these are the lines that have it.

• worked/src/BurstCeiling.java 72 lines
• worked/src/ClientCounts.java 64 lines
• worked/src/Drain.java 98 lines
• worked/src/DualGate.java 186 lines
• worked/src/Main.java 322 lines

worked/src/BurstCeiling.java72 lines

// BurstCeiling.java
//
// One rule about one field: at most `limit` admissions. Both classes answer the same question
// from 200 threads at once, and both compile clean under -Xlint:all. Main measures which one
// keeps the ceiling. The difference is not the field; it is whether the check and the charge
// are one operation or two.
import java.util.concurrent.atomic.AtomicInteger;

public final class BurstCeiling {

    private BurstCeiling() {}

    /**
     * The fix that fixes nothing. {@code volatile} makes every read of {@code used} fresh and
     * every write to it visible — and the admission decision is still a read (is there room?)
     * followed by a write (take it). Nothing joins them. Forty threads can read 24 before any
     * of them writes 25, and each of those forty is then entitled to admit.
     *
     * <p>Main, crowd of 200 against a limit of 20000, measured over fourteen runs: admitted
     * between 20028 and 21553, never the limit. The keyword changed which value each thread
     * saw. It never changed what each thread did next.
     */
    public static final class VolatileGuess {

        private final int limit;
        private volatile int used;

        public VolatileGuess(int limit) {
            this.limit = limit;
        }

        public boolean tryAcquire() {
            if (used < limit) {   // every thread can pass this line before any of them
                used++;           // reaches this one — and ++ is itself a read then a write
                return true;
            }
            return false;
        }

        public int used() {
            return used;
        }
    }

    /**
     * One field, one rule that mentions only that field: a conditional increment. So the whole
     * decision fits in one atomic call. {@code getAndUpdate} retries a compare-and-set until the
     * update applies to an unchanged value, and returns what it replaced — admitted exactly when
     * the value replaced was under the limit.
     *
     * <p>Two consequences worth stating. Exactly {@code limit} of any crowd admit, on every run,
     * as a guarantee rather than an observation. And {@code used} can never exceed {@code limit},
     * so a reader of {@link #used()} is never shown a count the rule forbids.
     */
    public static final class OneAtomic {

        private final int limit;
        private final AtomicInteger used = new AtomicInteger();

        public OneAtomic(int limit) {
            this.limit = limit;
        }

        public boolean tryAcquire() {
            return used.getAndUpdate(u -> u < limit ? u + 1 : u) < limit;
        }

        public int used() {
            return used.get();
        }
    }
}

worked/src/ClientCounts.java64 lines

// ClientCounts.java
//
// A denial tally per client key, the shape corpus/rate-limiter's curveball 03 ships for real:
// ConcurrentHashMap<key, AtomicLong>. The map handles concurrent access to its entries; the
// AtomicLong handles the increment. What neither handles is the seam between two calls — and
// "is this client registered yet?" followed by "register it" is two calls.
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

public final class ClientCounts {

    private ClientCounts() {}

    /**
     * Check-then-act on a {@link ConcurrentHashMap}. The {@code get} is atomic and the
     * {@code put} is atomic, and the null check between them is where every other thread's
     * {@code get} also lands. Two hundred threads hitting a fresh key all see {@code null},
     * all construct a counter, and all {@code put} it — last write wins, and every increment
     * made against a replaced counter vanishes with it.
     *
     * <p>The word {@code Concurrent} in the field's type is what lets this survive review.
     * The map keeps its own promises; it cannot know two of your calls were meant to be one.
     */
    public static final class Racy {

        private final Map<String, AtomicLong> counts = new ConcurrentHashMap<>();

        public void record(String client) {
            AtomicLong counter = counts.get(client);
            if (counter == null) {                 // the check
                counter = new AtomicLong();
                counts.put(client, counter);       // the act, and the window between them
            }
            counter.incrementAndGet();
        }

        public long count(String client) {
            AtomicLong counter = counts.get(client);
            return counter == null ? 0 : counter.get();
        }
    }

    /**
     * The same tally with the check and the act as one call. {@code computeIfAbsent} runs its
     * function while the map holds that bin's lock, so exactly one counter can ever exist per
     * key and every increment lands on it. The one-call family — {@code merge},
     * {@code compute}, {@code computeIfAbsent}, {@code computeIfPresent}, {@code putIfAbsent},
     * {@code replace} — is the answer whenever an update reads the map before writing it.
     */
    public static final class OneCall {

        private final Map<String, AtomicLong> counts = new ConcurrentHashMap<>();

        public void record(String client) {
            counts.computeIfAbsent(client, key -> new AtomicLong()).incrementAndGet();
        }

        public long count(String client) {
            AtomicLong counter = counts.get(client);
            return counter == null ? 0 : counter.get();
        }
    }
}

worked/src/Drain.java98 lines

// Drain.java
//
// The third kind of state: a flag one thread writes once and another thread only reads. This is
// the one shape volatile exists for, and the failure it prevents is not a lost update — nothing
// here is ever miscounted. The failure is a shutdown that never happens.
//
// corpus/logger's curveball 01 runs asynchronous destinations on one background thread. Hand-roll
// that worker instead of using an ExecutorService and this flag is the first field you write.
import java.util.concurrent.CountDownLatch;

public final class Drain {

    private Drain() {}

    /**
     * The worker's loop reads {@code stopped}; nothing in the worker ever writes it. That makes
     * hoisting the read out of the loop a legal transformation for a plain field, and the JIT
     * takes it: the loop becomes {@code while (!false)} with the flag read once, before the loop.
     * {@link #close} then sets a field nobody will ever read again.
     *
     * <p>Measured in Main: {@code close(1500)} timed out on every run. The worker is a daemon
     * thread here purely so the demo's JVM can exit — a real appender's worker would not be,
     * and the process would hang at shutdown with nothing on any stack trace to say why.
     */
    public static final class PlainFlag {

        private boolean stopped;
        private long served;
        private final CountDownLatch running = new CountDownLatch(1);
        private final Thread worker = new Thread(() -> {
            long n = 0;
            running.countDown();
            while (!stopped) {
                n++;
            }
            served = n;
        }, "drain-plain");

        public void start() throws InterruptedException {
            worker.setDaemon(true);
            worker.start();
            running.await();
        }

        /** True when the worker actually exited inside the wait. */
        public boolean close(long waitMillis) throws InterruptedException {
            stopped = true;
            worker.join(waitMillis);
            return !worker.isAlive();
        }

        /** Read after a successful close only — Thread.join is the happens-before edge. */
        public long served() {
            return served;
        }
    }

    /**
     * One keyword. A write to a volatile field happens-before every later read of it, and the
     * read cannot be hoisted, so the loop re-reads the flag and exits. Note what did NOT need
     * protection: {@code served} is written by one thread and read only after {@code join},
     * so it is a plain {@code long} — the join edge is already the guarantee.
     *
     * <p>The shape that earns {@code volatile}, all three conditions at once: one writer, a
     * write that does not read the current value, and no other field that must change with it.
     * Miss any one and the answer is an atomic or a lock instead.
     */
    public static final class VolatileFlag {

        private volatile boolean stopped;
        private long served;
        private final CountDownLatch running = new CountDownLatch(1);
        private final Thread worker = new Thread(() -> {
            long n = 0;
            running.countDown();
            while (!stopped) {
                n++;
            }
            served = n;
        }, "drain-volatile");

        public void start() throws InterruptedException {
            worker.setDaemon(true);
            worker.start();
            running.await();
        }

        public boolean close(long waitMillis) throws InterruptedException {
            stopped = true;
            worker.join(waitMillis);
            return !worker.isAlive();
        }

        public long served() {
            return served;
        }
    }
}

worked/src/DualGate.java186 lines

// DualGate.java
//
// Two rules on one client, judged as one decision: a burst ceiling and an hourly quota. The
// contract is the rate limiter's: check every rule, then charge every rule, and a request any
// rule refuses charges nothing. That unit — several reads, then several writes, all-or-nothing —
// is the thing that has to be atomic. It spans two fields, so no per-field device can be it.
//
// Three versions. The first is what ships under time pressure. The second is the first with
// atomics sprinkled on, which shrinks the window without closing it. The third names the unit.
import java.util.concurrent.atomic.AtomicInteger;

public final class DualGate {

    private DualGate() {}

    /**
     * The answer to one request. The compact constructor refuses a negative headroom, the same
     * check {@code corpus/rate-limiter/contract/Decision.java} makes. That is invariant
     * placement (A5) paying for itself under concurrency: a torn count cannot be quietly
     * reported, because the value type it would travel in cannot exist. The rate limiter's own
     * stress tuning saw exactly this — {@code remaining cannot be negative, got -3} — from a
     * broken variant no test was aimed at.
     */
    public record Verdict(boolean allowed, int remainingHour) {

        public Verdict {
            if (remainingHour < 0) {
                throw new IllegalArgumentException("remainingHour cannot be negative, got " + remainingHour);
            }
        }
    }

    /**
     * What each rule has been charged, read as one value. All-or-nothing means the two numbers
     * are equal at every moment anyone can observe — a snapshot where they differ describes a
     * request that was charged to one rule and not the other, a state the contract says cannot
     * exist.
     */
    public record Charges(int burst, int hour) {

        public boolean consistent() {
            return burst == hour;
        }
    }

    /** One client's budget: two rules, one decision. */
    public interface Budget {

        Verdict tryAcquire();

        Charges charges();
    }

    /**
     * No protection at all: plain fields, check both, charge both. Passes every test that calls
     * it from one thread, which is every test most candidates write. Under Main's storm it
     * admits more than the two limits allow, and the two counters drift apart — the hourly rule
     * ends up charged for fewer requests than were admitted, because {@code hourUsed++} lost
     * updates. Both numbers it reports afterwards are fiction.
     */
    public static final class NoLock implements Budget {

        private final int burstLimit;
        private final int hourLimit;
        private int burstUsed;
        private int hourUsed;

        public NoLock(int burstLimit, int hourLimit) {
            this.burstLimit = burstLimit;
            this.hourLimit = hourLimit;
        }

        @Override
        public Verdict tryAcquire() {
            if (burstUsed < burstLimit && hourUsed < hourLimit) {   // the check
                burstUsed++;                                        // the charge — two plain
                hourUsed++;                                         // read-modify-writes
                return new Verdict(true, hourLimit - hourUsed);
            }
            return new Verdict(false, hourLimit - hourUsed);
        }

        @Override
        public Charges charges() {
            return new Charges(burstUsed, hourUsed);
        }
    }

    /**
     * The correction that feels sufficient: one {@link AtomicInteger} per counter. No increment
     * is ever lost now, each field on its own is exact — and the decision is still two reads
     * followed by two writes, with every other thread landing in between. Main observes the
     * consequence directly: snapshots in which the burst rule has been charged and the hourly
     * rule has not, thousands of them per storm. All-or-nothing is a rule about the pair, and
     * there is no way to update the pair, or even to read it, as one moment.
     *
     * <p>The residual over-admission at the exact boundary is real too, and rare — Main hammers
     * one budget repeatedly and reports how often it fired. A failure that fires this rarely is
     * not safety. It is a bug with good camouflage.
     *
     * <p>The mistake is not the choice of {@code AtomicInteger}. It is believing atomicity per
     * field adds up to atomicity across fields.
     */
    public static final class PerFieldAtomics implements Budget {

        private final int burstLimit;
        private final int hourLimit;
        private final AtomicInteger burstUsed = new AtomicInteger();
        private final AtomicInteger hourUsed = new AtomicInteger();

        public PerFieldAtomics(int burstLimit, int hourLimit) {
            this.burstLimit = burstLimit;
            this.hourLimit = hourLimit;
        }

        @Override
        public Verdict tryAcquire() {
            if (burstUsed.get() < burstLimit && hourUsed.get() < hourLimit) {  // the check
                burstUsed.incrementAndGet();                                   // the charge,
                hourUsed.incrementAndGet();                                    // two steps late
                return new Verdict(true, hourLimit - hourUsed.get());
            }
            return new Verdict(false, hourLimit - hourUsed.get());
        }

        @Override
        public Charges charges() {
            return new Charges(burstUsed.get(), hourUsed.get());  // two moments, not one
        }
    }

    /**
     * The sentence that scores in a design review, written as code: two counters change
     * together under one rule, so they share exactly one lock. {@code private}, so no caller
     * can take it; {@code final}, so two threads can never lock different objects while both
     * believe they guard these fields.
     *
     * <p>The fields inside are plain {@code int}s — no {@code volatile}, no atomic wrapper.
     * Releasing the monitor happens-before the next acquisition of it, so the lock is already
     * the visibility guarantee. Decorating the fields anyway would tell a reader you were not
     * sure which mechanism was doing the work.
     */
    public static final class Locked implements Budget {

        private final int burstLimit;
        private final int hourLimit;
        private final Object lock = new Object();
        private int burstUsed;
        private int hourUsed;

        public Locked(int burstLimit, int hourLimit) {
            this.burstLimit = burstLimit;
            this.hourLimit = hourLimit;
        }

        @Override
        public Verdict tryAcquire() {
            synchronized (lock) {
                if (burstUsed < burstLimit && hourUsed < hourLimit) {
                    burstUsed++;
                    hourUsed++;
                    return new Verdict(true, hourLimit - hourUsed);
                }
                return new Verdict(false, hourLimit - hourUsed);
            }
        }

        /**
         * A read that reports on locked state is bound by the same lock. Unguarded it would
         * compile, pass every quiet test, and be exactly the unsynchronized {@code get} that
         * {@code corpus/lru-cache}'s stress suite was built to catch.
         */
        @Override
        public Charges charges() {
            synchronized (lock) {
                return new Charges(burstUsed, hourUsed);
            }
        }

        public int remainingHour() {
            synchronized (lock) {
                return hourLimit - hourUsed;
            }
        }
    }
}

worked/src/Main.java322 lines

// Main.java
//
// Every claim in this lesson, measured. Run it:
//
//   ..\..\..\..\.toolchain\jdk-21\bin\javac.exe -Xlint:all -d out *.java
//   ..\..\..\..\.toolchain\jdk-21\bin\java.exe -cp out Main
//
// Four blocks, one per decision. The broken half of each block compiles clean and passes any
// single-threaded test you could write for it — which is why the numbers below exist at all.
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;

public final class Main {

    /** The crowd both graded stress suites in the corpus use. */
    private static final int CROWD = 200;

    private Main() {}

    public static void main(String[] args) throws Exception {
        System.out.printf("crowd = %d threads%n%n", CROWD);
        blockOneCeiling();
        blockTwoDualGate();
        blockThreeRegistration();
        blockFourStopFlag();
    }

    // --- 1. one field, one rule: the volatile guess against the one-call atomic ---------------

    /**
     * Every thread makes many attempts, and the limit sits halfway through the total, so the
     * ceiling is reached while all 200 threads are still hot. One attempt per thread saturates
     * the gate before the threads genuinely overlap, and the broken class prints the right
     * number — the same trap the corpus tuning logs record.
     */
    private static void blockOneCeiling() throws InterruptedException {
        int perThread = 200;
        int limit = CROWD * perThread / 2;

        // Each thread tallies its own admissions in a local and publishes once at the end.
        // An AtomicInteger bumped inside the loop would serialise the threads on the
        // instrument and hide the very race being measured.
        BurstCeiling.VolatileGuess guess = new BurstCeiling.VolatileGuess(limit);
        AtomicInteger guessAdmitted = new AtomicInteger();
        contend(CROWD, () -> {
            int mine = 0;
            for (int i = 0; i < perThread; i++) {
                if (guess.tryAcquire()) {
                    mine++;
                }
            }
            guessAdmitted.addAndGet(mine);
        });

        BurstCeiling.OneAtomic atomic = new BurstCeiling.OneAtomic(limit);
        AtomicInteger atomicAdmitted = new AtomicInteger();
        contend(CROWD, () -> {
            int mine = 0;
            for (int i = 0; i < perThread; i++) {
                if (atomic.tryAcquire()) {
                    mine++;
                }
            }
            atomicAdmitted.addAndGet(mine);
        });

        System.out.printf("--- 1. a ceiling of %d, %d threads, %d attempts each%n", limit, CROWD, perThread);
        System.out.printf("  volatile int, check then act   admitted %d  (over by %d)%n",
                guessAdmitted.get(), guessAdmitted.get() - limit);
        System.out.printf("  AtomicInteger.getAndUpdate     admitted %d  (over by %d)%n%n",
                atomicAdmitted.get(), atomicAdmitted.get() - limit);
    }

    // --- 2. two fields, one decision: per-field atomics against one lock ----------------------

    /**
     * One budget per client key, exactly as the rate limiter keeps one {@code KeyBudget} per
     * {@code ClientKey}. Fifty keys means fifty limit boundaries crossed per storm — a single
     * gate saturates in the first moments of a storm, while most of the crowd is still waking,
     * and a race that gets one chance per storm is a race a demo cannot rely on. The corpus
     * tuning logs (lru-cache, rate-limiter) record the same lesson as catch rates.
     */
    private static void blockTwoDualGate() throws InterruptedException {
        int keys = 50;
        int perKey = 2;      // attempts per thread per key: 400 arrive at each budget of 200
        int burst = 200;
        int hour = 300;

        System.out.printf("--- 2. burst %d + hourly %d per client, all-or-nothing, %d clients, "
                + "%d threads x %d attempts each%n", burst, hour, keys, CROWD, perKey);

        StormResult noLock = stormBudgets(() -> new DualGate.NoLock(burst, hour), keys, perKey);
        System.out.printf("  plain ints, no lock          admitted %d of %d allowed (over by %d); "
                + "hourly rules charged %d for %d admitted; torn snapshots seen: %d; "
                + "unbuildable Verdicts: %d%n",
                noLock.through(), keys * burst, noLock.through() - keys * burst,
                noLock.hourCharged(), noLock.through(), noLock.tornSnapshots(), noLock.unbuildable());

        StormResult atomics = stormBudgets(() -> new DualGate.PerFieldAtomics(burst, hour), keys, perKey);
        int atomicsBreaches = hammer(() -> new DualGate.PerFieldAtomics(10_000, 15_000));
        System.out.printf("  one AtomicInteger per field  admitted %d of %d allowed (over by %d); "
                + "torn snapshots seen: %d; boundary hammer breached %d of %d single-budget storms%n",
                atomics.through(), keys * burst, atomics.through() - keys * burst,
                atomics.tornSnapshots(), atomicsBreaches, HAMMER_STORMS);

        StormResult locked = stormBudgets(() -> new DualGate.Locked(burst, hour), keys, perKey);
        int lockedBreaches = hammer(() -> new DualGate.Locked(10_000, 15_000));
        System.out.printf("  one lock over the decision   admitted %d of %d allowed (over by %d); "
                + "torn snapshots seen: %d; boundary hammer breached %d of %d single-budget storms%n%n",
                locked.through(), keys * burst, locked.through() - keys * burst,
                locked.tornSnapshots(), lockedBreaches, HAMMER_STORMS);
    }

    private record StormResult(int through, int hourCharged, long tornSnapshots, int unbuildable) {}

    /**
     * One storm across many budgets, with an observer thread reading every budget's pair of
     * charges the whole time. All-or-nothing means the pair is equal at every observable moment,
     * so any snapshot where it is not is a request half-charged in public view.
     */
    private static StormResult stormBudgets(Supplier<DualGate.Budget> maker, int keys, int perKey)
            throws InterruptedException {
        DualGate.Budget[] gates = new DualGate.Budget[keys];
        for (int k = 0; k < keys; k++) {
            gates[k] = maker.get();
        }

        AtomicBoolean storming = new AtomicBoolean(true);
        AtomicLong torn = new AtomicLong();
        Thread observer = new Thread(() -> {
            long seen = 0;
            while (storming.get()) {
                for (DualGate.Budget gate : gates) {
                    if (!gate.charges().consistent()) {
                        seen++;
                    }
                }
            }
            torn.set(seen);
        }, "charges-observer");
        observer.start();

        AtomicInteger admitted = new AtomicInteger();
        AtomicInteger unbuildable = new AtomicInteger();
        contend(CROWD, () -> {
            int mine = 0;
            int failed = 0;
            for (DualGate.Budget gate : gates) {
                for (int i = 0; i < perKey; i++) {
                    try {
                        if (gate.tryAcquire().allowed()) {
                            mine++;
                        }
                    } catch (IllegalArgumentException tornCount) {
                        // Verdict's constructor refused a negative headroom mid-race. The
                        // request was already charged; only the report of it could not be built.
                        failed++;
                    }
                }
            }
            admitted.addAndGet(mine);
            unbuildable.addAndGet(failed);
        });
        storming.set(false);
        observer.join();

        int hourCharged = 0;
        for (DualGate.Budget gate : gates) {
            hourCharged += gate.charges().hour();
        }
        // A Verdict that threw was thrown AFTER the charge landed, so it still went through.
        return new StormResult(admitted.get() + unbuildable.get(), hourCharged, torn.get(),
                unbuildable.get());
    }

    private static final int HAMMER_STORMS = 30;

    /**
     * The residual boundary race: one budget, its limit sitting mid-storm, hammered repeatedly.
     * Reports how many storms admitted more than the burst limit. For the atomics variant this
     * fires rarely — single digits of storms, over by single digits, and on some whole runs not
     * at all. Read that as camouflage, not safety: the corpus tuning logs treat a bug that
     * escapes a small number of attempts as a bug that needs more attempts.
     */
    private static int hammer(Supplier<DualGate.Budget> maker) throws InterruptedException {
        int perThread = 100;
        int burstLimit = 10_000;
        int breaches = 0;
        for (int storm = 0; storm < HAMMER_STORMS; storm++) {
            DualGate.Budget gate = maker.get();
            AtomicInteger admitted = new AtomicInteger();
            AtomicInteger unbuildable = new AtomicInteger();
            contend(CROWD, () -> {
                int mine = 0;
                int failed = 0;
                for (int i = 0; i < perThread; i++) {
                    try {
                        if (gate.tryAcquire().allowed()) {
                            mine++;
                        }
                    } catch (IllegalArgumentException tornCount) {
                        failed++;
                    }
                }
                admitted.addAndGet(mine);
                unbuildable.addAndGet(failed);
            });
            if (admitted.get() + unbuildable.get() > burstLimit) {
                breaches++;
            }
        }
        return breaches;
    }

    // --- 3. check-then-act on a ConcurrentHashMap: registration --------------------------------

    private static void blockThreeRegistration() throws InterruptedException {
        int rounds = 20;
        int perThread = 50;
        long expected = (long) rounds * CROWD * perThread;

        ClientCounts.Racy racy = new ClientCounts.Racy();
        long racyTotal = 0;
        for (int round = 0; round < rounds; round++) {
            String key = "client-" + round;
            contend(CROWD, () -> {
                for (int i = 0; i < perThread; i++) {
                    racy.record(key);
                }
            });
            racyTotal += racy.count(key);
        }

        ClientCounts.OneCall oneCall = new ClientCounts.OneCall();
        long oneCallTotal = 0;
        for (int round = 0; round < rounds; round++) {
            String key = "client-" + round;
            contend(CROWD, () -> {
                for (int i = 0; i < perThread; i++) {
                    oneCall.record(key);
                }
            });
            oneCallTotal += oneCall.count(key);
        }

        System.out.printf("--- 3. registering a counter on a ConcurrentHashMap: %d fresh keys, "
                + "%d threads x %d records each%n", rounds, CROWD, perThread);
        System.out.printf("  get, null-check, put            counted %d of %d  (lost %d)%n",
                racyTotal, expected, expected - racyTotal);
        System.out.printf("  computeIfAbsent                 counted %d of %d  (lost %d)%n%n",
                oneCallTotal, expected, expected - oneCallTotal);
    }

    // --- 4. the stop flag: a visibility failure with nothing miscounted ------------------------

    private static void blockFourStopFlag() throws InterruptedException {
        Drain.PlainFlag plain = new Drain.PlainFlag();
        plain.start();
        Thread.sleep(100);  // let the JIT meet the loop before the flag flips
        boolean plainClosed = plain.close(1500);

        Drain.VolatileFlag guarded = new Drain.VolatileFlag();
        guarded.start();
        Thread.sleep(100);
        boolean guardedClosed = guarded.close(1500);

        System.out.printf("--- 4. close() sets a stop flag and waits up to 1500ms for the worker%n");
        System.out.printf("  plain boolean      worker exited: %b%n", plainClosed);
        System.out.printf("  volatile boolean   worker exited: %b  (served %d polls before stopping)%n",
                guardedClosed, guarded.served());
    }

    // --- coordination ---------------------------------------------------------------------------

    /**
     * Every task released from one latch, so the threads collide instead of drifting past each
     * other. Start 200 threads without this and the early ones finish before the late ones
     * begin — and every broken class above prints the right number while still being broken.
     */
    private static void contend(int threads, Runnable task) throws InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(threads);
        CountDownLatch ready = new CountDownLatch(threads);
        CountDownLatch go = new CountDownLatch(1);
        CountDownLatch done = new CountDownLatch(threads);
        AtomicLong failures = new AtomicLong();
        try {
            for (int i = 0; i < threads; i++) {
                pool.submit(() -> {
                    ready.countDown();
                    try {
                        go.await();
                        task.run();
                    } catch (InterruptedException interrupted) {
                        Thread.currentThread().interrupt();
                    } catch (RuntimeException unexpected) {
                        failures.incrementAndGet();
                    } finally {
                        done.countDown();
                    }
                });
            }
            if (!ready.await(30, TimeUnit.SECONDS)) {
                throw new IllegalStateException("threads did not reach the start line");
            }
            go.countDown();
            if (!done.await(30, TimeUnit.SECONDS)) {
                throw new IllegalStateException("tasks did not finish in 30s — a lock held too long, or a deadlock");
            }
        } finally {
            pool.shutdownNow();
        }
        if (failures.get() > 0) {
            throw new IllegalStateException(failures.get() + " task(s) threw unexpectedly");
        }
    }
}