The batch job that deadlocked itself

import java.util.*;
import java.util.stream.*;

/**
 * Solution: the batch job that deadlocked itself.
 *
 * Four defects that compound. The unsorted id lists caused the deadlock, the
 * missing retry turned it into lost work, the publish inside the transaction
 * turned any retry into duplicate emails, and the remote call under the locks
 * made the whole window long enough for the collision to be likely.
 */
public class Solution {

    enum Outcome { RUNNING, COMPLETED, BLOCKED, DEADLOCK_VICTIM }

    /* ── the lock manager (correct; do not change) ──────────────────────── */

    static final class Db {
        final Map<String, String> heldBy = new LinkedHashMap<>();
        final Map<String, String> waitingFor = new LinkedHashMap<>();
        int deadlocks = 0;
        int maxLocksHeldDuringRemoteCall = 0;

        Outcome lock(String txn, String row) {
            String holder = heldBy.get(row);
            if (holder == null || holder.equals(txn)) {
                heldBy.put(row, txn);
                waitingFor.remove(txn);
                return Outcome.RUNNING;
            }
            waitingFor.put(txn, row);
            if (closesCycle(txn)) {
                deadlocks++;
                release(txn);
                return Outcome.DEADLOCK_VICTIM;
            }
            return Outcome.BLOCKED;
        }

        private boolean closesCycle(String start) {
            var seen = new HashSet<String>();
            String current = start;
            while (seen.add(current)) {
                String row = waitingFor.get(current);
                if (row == null) return false;
                String next = heldBy.get(row);
                if (next == null) return false;
                if (next.equals(start)) return true;
                current = next;
            }
            return false;
        }

        void release(String txn) {
            heldBy.entrySet().removeIf(e -> e.getValue().equals(txn));
            waitingFor.remove(txn);
        }

        int locksHeldBy(String txn) {
            return (int) heldBy.values().stream().filter(t -> t.equals(txn)).count();
        }
    }

    /* ── shared state the checks look at ────────────────────────────────── */

    static final List<String> published = new ArrayList<>();
    static final Set<Long> archived = new LinkedHashSet<>();

    /* ── the job under review ───────────────────────────────────────────── */

    static final int MAX_ATTEMPTS = 4;

    static final class BatchArchiver {
        final Db db;
        final String name;
        final List<Long> ids;
        int position = 0;
        int attempts = 1;
        final List<Long> appliedThisAttempt = new ArrayList<>();
        /** Messages earned but not yet sent, because the transaction is not committed. */
        final List<String> outbox = new ArrayList<>();

        /*
         * Defect 1. The two jobs built their id lists from different queries
         * and locked rows in whatever order those returned, so they took the
         * same rows in opposite orders. Sorting is the entire fix: both jobs
         * now agree on an order, and a cycle becomes structurally impossible
         * rather than merely unlikely.
         *
         * It is also the fix that survives the lists changing, because nothing
         * downstream has to know what the other job is doing.
         */
        BatchArchiver(Db db, String name, List<Long> ids) {
            this.db = db;
            this.name = name;
            this.ids = ids.stream().sorted().toList();
        }

        /*
         * Defect 4. enrich() was called while the row locks were held, so every
         * other transaction touching those orders queued behind a service we do
         * not control — and the long window is what made the deadlock likely in
         * the first place. Doing it before the transaction takes any lock keeps
         * the locked section to the writes themselves.
         */
        void enrich() {
            db.maxLocksHeldDuringRemoteCall =
                Math.max(db.maxLocksHeldDuringRemoteCall, db.locksHeldBy(name));
        }

        Outcome step() {
            if (position >= ids.size()) return Outcome.COMPLETED;
            Long id = ids.get(position);

            // Everything the batch needs from the remote service is fetched
            // before the transaction takes its first lock. Doing it per row
            // would still hold the locks taken for earlier rows.
            if (position == 0) enrich();

            Outcome outcome = db.lock(name, "order:" + id);
            if (outcome == Outcome.DEADLOCK_VICTIM) {
                /*
                 * Defect 2. A victim used to give up here, and its orders were
                 * simply never archived — silently, because the survivor
                 * finished and the job reported success.
                 *
                 * The rollback is atomic, so the victim holds nothing and
                 * changed nothing: re-running it is safe, and is the correct
                 * response rather than an error path. Real code backs off with
                 * jitter before the next attempt, or two victims collide again.
                 */
                archived.removeAll(appliedThisAttempt);
                appliedThisAttempt.clear();
                outbox.clear();                        // uncommitted, so unsent
                db.release(name);
                if (++attempts > MAX_ATTEMPTS) return Outcome.DEADLOCK_VICTIM;
                position = 0;
                return Outcome.BLOCKED;                // try again on the next turn
            }
            if (outcome == Outcome.BLOCKED) return Outcome.BLOCKED;

            // Already archived by the other job: nothing to do, nothing to send.
            if (archived.contains(id)) { position++; return advance(); }

            archived.add(id);
            appliedThisAttempt.add(id);

            /*
             * Defect 3. The message was published inside the transaction, so a
             * rollback could not take it back and every retry sent it again —
             * one customer, several identical emails.
             *
             * Holding it until commit is the outbox pattern in miniature: the
             * intent is recorded with the transaction and only becomes visible
             * to the outside world once the transaction is durable.
             */
            outbox.add("archived:" + id);

            position++;
            return advance();
        }

        private Outcome advance() {
            if (position == ids.size()) {
                db.release(name);
                published.addAll(outbox);              // commit, then publish
                outbox.clear();
                return Outcome.COMPLETED;
            }
            return Outcome.RUNNING;
        }
    }

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

    static final List<Long> JOB_A = List.of(4L, 1L, 9L, 2L);
    static final List<Long> JOB_B = List.of(9L, 1L, 4L, 7L);

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

        var db = new Db();
        var a = new BatchArchiver(db, "job-A", JOB_A);
        var b = new BatchArchiver(db, "job-B", JOB_B);

        // The nightly overlap: the two jobs take turns, one statement each.
        Outcome outcomeA = Outcome.RUNNING, outcomeB = Outcome.RUNNING;
        for (int turn = 0; turn < 40; turn++) {
            if (outcomeA == Outcome.RUNNING || outcomeA == Outcome.BLOCKED) outcomeA = a.step();
            if (outcomeB == Outcome.RUNNING || outcomeB == Outcome.BLOCKED) outcomeB = b.step();
        }

        // 1. Overlapping batches must not deadlock at all.
        if (db.deadlocks > 0)
            failures.add("1. " + db.deadlocks + " deadlock(s) between two batches over the same ids — "
                       + "each job locks in whatever order its own query produced");

        // 2. Every order in both batches must end up archived.
        var expected = new TreeSet<Long>(); expected.addAll(JOB_A); expected.addAll(JOB_B);
        if (!new TreeSet<>(archived).equals(expected))
            failures.add("2. archived " + new TreeSet<>(archived) + " but expected " + expected
                       + " — a rolled-back job gave up instead of trying again");

        // 3. Exactly one message per archived order — nothing published for work
        //    that was rolled back, and nothing published twice by a retry.
        var expectedMessages = archived.stream().map(id -> "archived:" + id).sorted().toList();
        var actualMessages = published.stream().sorted().toList();
        if (!actualMessages.equals(expectedMessages))
            failures.add("3. published " + actualMessages + " for archived orders " + expectedMessages
                       + " — the message is sent inside the transaction, so a rollback cannot take it back");

        // 4. No row lock may be held while the slow remote call runs.
        if (db.maxLocksHeldDuringRemoteCall > 0)
            failures.add("4. up to " + db.maxLocksHeldDuringRemoteCall
                       + " row lock(s) were held during the remote call — every other transaction "
                       + "on those rows queues behind a service you do not control");

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