The balance that lost a top-up

import java.util.*;
import java.util.function.*;

/**
 * Solution: the balance that lost a top-up.
 *
 * Four defects, and only one of them is fixed by choosing a different
 * isolation level. The other three need a retry, a lock, or a different
 * operation — which is the point. "Raise the isolation level" is the answer to
 * exactly one of these problems.
 */
public class Solution {

    enum Level { READ_COMMITTED, REPEATABLE_READ, SERIALIZABLE }

    /* ── the store (correct; do not change) ─────────────────────────────── */

    static final class Version {
        final String key; final int value; final long writer;
        long committedAt = Long.MAX_VALUE;
        Version(String key, int value, long writer) { this.key = key; this.value = value; this.writer = writer; }
    }

    static final class Store {
        final List<Version> versions = new ArrayList<>();
        long clock = 0;
        /** Forces the next N commits to be refused, the way contention would. */
        int refuseNextCommits = 0;

        Store seed(String key, int value) {
            var v = new Version(key, value, 0);
            v.committedAt = ++clock;
            versions.add(v);
            return this;
        }

        Txn begin(Level level) { return new Txn(this, level, ++clock); }

        List<Version> history(String key) {
            var out = new ArrayList<Version>();
            for (int i = versions.size() - 1; i >= 0; i--)
                if (versions.get(i).key.equals(key)) out.add(versions.get(i));
            return out;
        }

        int committedValue(String key) {
            for (Version v : history(key))
                if (v.committedAt != Long.MAX_VALUE) return v.value;
            return 0;
        }
    }

    static final class Txn {
        final Store store; final Level level; final long startedAt;
        final List<Version> mine = new ArrayList<>();
        final Set<String> readSet = new LinkedHashSet<>();
        boolean aborted;

        Txn(Store store, Level level, long startedAt) {
            this.store = store; this.level = level; this.startedAt = startedAt;
        }

        Integer read(String key) {
            readSet.add(key);
            for (Version v : store.history(key)) {
                if (v.writer == startedAt) return v.value;
                boolean visible = level == Level.READ_COMMITTED
                    ? v.committedAt <= store.clock
                    : v.committedAt < startedAt;
                if (visible) return v.value;
            }
            return null;
        }

        List<String> readRange(String prefix) {
            var found = new TreeSet<String>();
            for (Version v : store.versions) {
                if (!v.key.startsWith(prefix)) continue;
                boolean visible = level == Level.READ_COMMITTED
                    ? v.committedAt <= store.clock
                    : v.committedAt < startedAt;
                if (visible) found.add(v.key);
            }
            return new ArrayList<>(found);
        }

        Txn write(String key, int value) {
            var v = new Version(key, value, startedAt);
            store.versions.add(v);
            mine.add(v);
            return this;
        }

        /** Returns false when the transaction is refused and must be retried. */
        boolean commit() {
            if (store.refuseNextCommits > 0) { store.refuseNextCommits--; aborted = true; return false; }
            if (level == Level.SERIALIZABLE) {
                for (String key : readSet)
                    for (Version v : store.history(key))
                        if (v.writer != startedAt && v.committedAt != Long.MAX_VALUE
                            && v.committedAt > startedAt) { aborted = true; return false; }
            }
            long at = ++store.clock;
            for (Version v : mine) v.committedAt = at;
            return true;
        }
    }

    /* ── the service under review ───────────────────────────────────────── */

    static final int MAX_ATTEMPTS = 5;

    static final class AccountService {
        final Store store;
        AccountService(Store store) { this.store = store; }

        /*
         * Defects 1 and 3. The credit read a balance, let another transaction
         * commit, then wrote a value computed from the stale read — a lost
         * update, which read committed permits on every database. And the
         * return value of commit() was discarded, so a refused transaction was
         * reported to the caller as a completed one.
         *
         * Serializable turns the silent overwrite into a refusal, and the
         * retry is what turns the refusal back into completed work. Neither
         * half is sufficient alone: without the retry this is still broken,
         * just noisily instead of quietly.
         */
        void credit(String account, int amount, Runnable betweenReadAndWrite) {
            for (int attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
                var txn = store.begin(Level.SERIALIZABLE);
                int current = txn.read(account);
                if (attempt == 1) betweenReadAndWrite.run();   // the conflict only lands once
                txn.write(account, current + amount);
                if (txn.commit()) return;
                // Real code backs off with jitter here before trying again.
            }
            throw new IllegalStateException("could not credit " + account + " after " + MAX_ATTEMPTS);
        }

        /*
         * Defect 2. Two reads in one statement at read committed can see two
         * different committed values, so the header disagreed with the detail.
         * Nothing was corrupt — the reads simply happened at different
         * instants.
         *
         * This is the ONE defect here that an isolation level fixes, because
         * the problem genuinely is read consistency. A snapshot taken at the
         * start makes every read in the transaction agree.
         */
        List<Integer> statement(String account, Runnable betweenReads) {
            var txn = store.begin(Level.REPEATABLE_READ);
            int header = txn.read(account);
            betweenReads.run();                       // the world moves on
            int detail = txn.read(account);
            txn.commit();
            return List.of(header, detail);
        }

        /*
         * Defect 4. max + 1 is a read-modify-write over a RANGE, so two
         * transactions reading the same snapshot both computed the same next
         * number. A row lock could not have helped: the row being protected
         * did not exist yet, which is what makes this a phantom rather than a
         * lost update.
         *
         * Serializable plus a retry closes it here. In a real schema the
         * better answer is a sequence or an identity column, and a UNIQUE
         * constraint on the number regardless — so that if this reasoning is
         * ever wrong, the database refuses the duplicate instead of storing it.
         */
        int nextInvoiceNumber(Runnable betweenReadAndWrite) {
            for (int attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
                var txn = store.begin(Level.SERIALIZABLE);
                int max = 0;
                for (String key : txn.readRange("invoice:")) {
                    max = Math.max(max, Integer.parseInt(key.substring("invoice:".length())));
                    txn.readSet.add(key);              // the range is what we depend on
                }
                txn.readSet.add("invoice:" + (max + 1));
                if (attempt == 1) betweenReadAndWrite.run();
                int next = max + 1;
                txn.write("invoice:" + next, 1);
                if (txn.commit()) return next;
            }
            throw new IllegalStateException("could not allocate an invoice number");
        }
    }

    /* ── checks ─────────────────────────────────────────────────────────── */

    public static void main(String[] args) {
        List<String> failures = new ArrayList<>();

        // 1. Two concurrent credits of 50 on a balance of 100 must leave 200.
        {
            var store = new Store().seed("alice", 100);
            var service = new AccountService(store);
            // A second credit runs to completion between the first one's read
            // and its write — the overlap two support agents produce by hand.
            service.credit("alice", 50, () -> service.credit("alice", 50, () -> {}));
            int viaService = store.committedValue("alice");
            if (viaService != 200)
                failures.add("1. two credits of 50 on 100 left " + viaService
                           + " — one was overwritten, and both transactions reported success");
        }

        // 2. Two reads inside one statement must agree with each other.
        {
            var store = new Store().seed("bob", 100);
            var service = new AccountService(store);
            var reads = service.statement("bob", () -> {
                var other = store.begin(Level.READ_COMMITTED);
                other.write("bob", 999);
                other.commit();
            });
            if (!reads.get(0).equals(reads.get(1)))
                failures.add("2. one statement read " + reads.get(0) + " then " + reads.get(1)
                           + " for the same account — the header will not match the detail");
        }

        // 3. A refused commit must be retried, not silently dropped. Under
        //    contention a database refuses commits routinely; the application
        //    is obliged to run the transaction again.
        {
            var store = new Store().seed("carol", 100);
            var service = new AccountService(store);
            store.refuseNextCommits = 1;              // the first attempt loses the race
            service.credit("carol", 75, () -> {});
            int balance = store.committedValue("carol");
            if (balance != 175)
                failures.add("3. after one refused commit the balance is " + balance
                           + ", not 175 — the credit was reported as done and never applied");
        }

        // 4. Concurrent invoice numbering must not produce a duplicate.
        {
            var store = new Store().seed("invoice:1", 1).seed("invoice:2", 1);
            var service = new AccountService(store);
            var issued = new ArrayList<Integer>();
            issued.add(service.nextInvoiceNumber(() -> {
                var other = new AccountService(store);
                issued.add(other.nextInvoiceNumber(() -> {}));
            }));
            if (new HashSet<>(issued).size() != issued.size())
                failures.add("4. invoice numbers " + issued + " contain a duplicate — "
                           + "max + 1 was computed twice from the same snapshot");
        }

        /* ── report ─────────────────────────────────────────────────────── */
        if (failures.isEmpty()) {
            System.out.println("PASS");
        } else {
            failures.forEach(f -> System.out.println("  " + f));
            System.out.println("FAIL");
        }
    }
}