W11 코드 뒤풀이 · 동시성은 숫자와 경계로 확인한다

package com.example.financialcore;

import org.junit.jupiter.api.Test;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

class ConcurrencyHarnessTest {
    record RunResult(int success, int businessFailure, int technicalFailure, int observedMaxActive) {}
    record Sample(String variant, double millis) {}
    record Stats(double median, double nearestRankP95) {}

    @Test
    void fixed_pool_bounds_observed_activity_and_accounts_for_all_tasks() throws Exception {
        for (int workers : List.of(10, 50, 200)) {
            RunResult result = run(workers, 1_000, Math.min(workers, 1_000), Duration.ofSeconds(10));
            assertThat(result.observedMaxActive()).isBetween(1, workers);
            assertThat(result.success() + result.businessFailure() + result.technicalFailure())
                .isEqualTo(1_000);
            assertThat(result.technicalFailure()).isZero();
        }
    }

    @Test
    void ready_timeout_is_a_failure_and_never_a_green_run() {
        assertThatThrownBy(() -> run(2, 2, 3, Duration.ofMillis(100)))
            .isInstanceOf(IllegalStateException.class)
            .hasMessageContaining("ready timeout");
    }

    @Test
    void even_median_and_nearest_rank_p95_match_the_independent_vector() {
        List<Double> values = IntStream.rangeClosed(1, 30).mapToObj(i -> (double) i).toList();
        Stats stats = statistics(values);
        assertThat(stats.median()).isEqualTo(15.5);
        assertThat(stats.nearestRankP95()).isEqualTo(29.0);
    }

    @Test
    void benchmark_contract_requires_exactly_before_and_after_with_thirty_rows_each() {
        List<Sample> valid = new ArrayList<>();
        for (String variant : List.of("before", "after")) {
            for (int value = 1; value <= 30; value++) valid.add(new Sample(variant, value));
        }
        assertThat(validateAndSummarize(valid)).containsOnlyKeys("before", "after");

        assertThatThrownBy(() -> validateAndSummarize(valid.subList(0, 59)))
            .isInstanceOf(IllegalArgumentException.class);
        List<Sample> extra = new ArrayList<>(valid);
        extra.add(new Sample("other", 1));
        assertThatThrownBy(() -> validateAndSummarize(extra))
            .isInstanceOf(IllegalArgumentException.class);
        assertThatThrownBy(() -> validateAndSummarize(List.of()))
            .isInstanceOf(IllegalArgumentException.class);
    }

    static RunResult run(
        int workers, int tasks, int readyTarget, Duration readyTimeout
    ) throws Exception {
        if (workers <= 0 || tasks <= 0 || readyTarget <= 0) throw new IllegalArgumentException();
        var ready = new CountDownLatch(readyTarget);
        var start = new CountDownLatch(1);
        var active = new AtomicInteger();
        var maxActive = new AtomicInteger();
        var pool = Executors.newFixedThreadPool(workers);
        List<Future<String>> futures = new ArrayList<>();
        try {
            for (int i = 0; i < tasks; i++) {
                futures.add(pool.submit(() -> {
                    ready.countDown();
                    start.await();
                    int now = active.incrementAndGet();
                    maxActive.accumulateAndGet(now, Math::max);
                    try {
                        LockSupport.parkNanos(100_000L);
                        return "SUCCESS";
                    } finally {
                        active.decrementAndGet();
                    }
                }));
            }
            if (!ready.await(readyTimeout.toMillis(), TimeUnit.MILLISECONDS)) {
                throw new IllegalStateException("ready timeout");
            }
            start.countDown();
            int success = 0;
            int business = 0;
            int technical = 0;
            for (Future<String> future : futures) {
                try {
                    if ("SUCCESS".equals(future.get(30, TimeUnit.SECONDS))) success++;
                    else business++;
                } catch (Exception failure) {
                    technical++;
                }
            }
            return new RunResult(success, business, technical, maxActive.get());
        } finally {
            start.countDown();
            pool.shutdownNow();
        }
    }

    static Map<String, Stats> validateAndSummarize(List<Sample> samples) {
        Map<String, List<Double>> byVariant = samples.stream().collect(Collectors.groupingBy(
            Sample::variant, Collectors.mapping(Sample::millis, Collectors.toList())
        ));
        if (!byVariant.keySet().equals(Set.of("before", "after"))) {
            throw new IllegalArgumentException("variants must be exactly before and after");
        }
        if (byVariant.values().stream().anyMatch(values -> values.size() != 30 || values.isEmpty())) {
            throw new IllegalArgumentException("each variant must contain exactly 30 rows");
        }
        return byVariant.entrySet().stream().collect(Collectors.toMap(
            Map.Entry::getKey, entry -> statistics(entry.getValue())
        ));
    }

    static Stats statistics(List<Double> source) {
        if (source.isEmpty()) throw new IllegalArgumentException("empty samples");
        List<Double> values = source.stream().sorted(Comparator.naturalOrder()).toList();
        int n = values.size();
        double median = n % 2 == 0
            ? (values.get(n / 2 - 1) + values.get(n / 2)) / 2.0
            : values.get(n / 2);
        int nearestRank = Math.max(1, (int) Math.ceil(n * 0.95));
        return new Stats(median, values.get(nearestRank - 1));
    }
}

package com.example.financialcore.account;

import com.example.financialcore.PostgresIntegrationTestSupport;
import com.example.financialcore.api.BusinessException;
import com.example.financialcore.api.ErrorCode;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.jdbc.core.simple.JdbcClient;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest
class ConcurrentWithdraw20IT extends PostgresIntegrationTestSupport {
    @Autowired JdbcClient jdbc;
    @Autowired AccountOpeningService openings;
    @Autowired AccountRepository accounts;
    @Autowired WithdrawalService withdrawals;

    Account account;

    @BeforeEach
    void setUp() {
        jdbc.sql("TRUNCATE idempotency_request,ledger_entry,business_tx,account RESTART IDENTITY CASCADE")
            .update();
        account = openings.open("customer-1", "WITHDRAW-20", 10_000);
    }

    @Test
    void twentyConcurrentWithdrawalsHaveTenSuccessTenBusinessFailureAndNoTechnicalFailure() throws Exception {
        int tasks = 20;
        var ready = new CountDownLatch(tasks);
        var start = new CountDownLatch(1);
        var pool = Executors.newFixedThreadPool(tasks);
        try {
            List<Future<String>> futures = new ArrayList<>();
            for (int i = 0; i < tasks; i++) {
                int sequence = i;
                futures.add(pool.submit(() -> {
                    ready.countDown();
                    start.await();
                    try {
                        withdrawals.withdraw("customer-1", account.getId(), 1_000,
                            "withdraw-" + sequence);
                        return "SUCCESS";
                    } catch (BusinessException failure) {
                        return failure.code() == ErrorCode.INSUFFICIENT_BALANCE
                            ? "BUSINESS_FAILURE" : "TECHNICAL_FAILURE";
                    } catch (RuntimeException failure) {
                        return "TECHNICAL_FAILURE";
                    }
                }));
            }
            assertThat(ready.await(10, TimeUnit.SECONDS)).isTrue();
            start.countDown();
            List<String> outcomes = new ArrayList<>();
            for (Future<String> future : futures) outcomes.add(future.get(30, TimeUnit.SECONDS));
            assertThat(outcomes).as("W11D2_RED_EXPECTED_ATOMIC_WITHDRAWAL")
                .filteredOn("SUCCESS"::equals).hasSize(10);
            assertThat(outcomes).filteredOn("BUSINESS_FAILURE"::equals).hasSize(10);
            assertThat(outcomes).filteredOn("TECHNICAL_FAILURE"::equals).isEmpty();
            assertThat(accounts.findById(account.getId()).orElseThrow().getBalance()).isZero();
            assertThat(jdbc.sql("SELECT COUNT(*) FROM ledger_entry WHERE entry_type='WITHDRAWAL'")
                .query(Long.class).single()).isEqualTo(10);
        } finally {
            start.countDown();
            pool.shutdownNow();
        }
    }
}

package com.example.financialcore.account;

import com.example.financialcore.api.BusinessException;
import com.example.financialcore.api.ErrorCode;
import com.example.financialcore.ledger.BusinessTransaction;
import com.example.financialcore.ledger.BusinessTransactionRepository;
import com.example.financialcore.ledger.LedgerEntry;
import com.example.financialcore.ledger.LedgerEntryRepository;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;

@Service
public class WithdrawalService {
    private final AccountRepository accounts;
    private final BusinessTransactionRepository transactions;
    private final LedgerEntryRepository ledger;

    public WithdrawalService(
        AccountRepository accounts,
        BusinessTransactionRepository transactions,
        LedgerEntryRepository ledger
    ) {
        this.accounts = accounts;
        this.transactions = transactions;
        this.ledger = ledger;
    }

    @Transactional
    public long withdraw(String actorId, long accountId, long amount, String requestId) {
        if (amount <= 0) throw new BusinessException(ErrorCode.INVALID_REQUEST, "amount must be positive");
        Account account = accounts.findOneForUpdate(accountId)
            .orElseThrow(() -> new BusinessException(ErrorCode.ACCOUNT_NOT_FOUND, "account not found"));
        if (!account.getOwnerId().equals(actorId)) {
            throw new BusinessException(ErrorCode.ACCESS_DENIED, "access denied");
        }
        account.withdraw(amount);
        Instant now = Instant.now();
        BusinessTransaction tx = transactions.save(BusinessTransaction.completedWithdrawal(
            "WITHDRAW:" + accountId + ":" + requestId, now));
        ledger.save(LedgerEntry.withdrawal(tx, account, amount, now));
        return account.getBalance();
    }
}

package com.example.financialcore.transfer;

import com.example.financialcore.PostgresIntegrationTestSupport;
import com.example.financialcore.account.Account;
import com.example.financialcore.account.AccountOpeningService;
import com.example.financialcore.account.AccountRepository;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.jdbc.core.simple.JdbcClient;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest
class SortedLockTransferIT extends PostgresIntegrationTestSupport {
    @Autowired JdbcClient jdbc;
    @Autowired AccountOpeningService openings;
    @Autowired AccountRepository accounts;
    @Autowired TransferService transfers;
    Account first;
    Account second;

    @BeforeEach
    void setUp() {
        jdbc.sql("TRUNCATE idempotency_request,ledger_entry,business_tx,account RESTART IDENTITY CASCADE")
            .update();
        first = openings.open("customer-1", "SORT-A", 10_000);
        second = openings.open("customer-1", "SORT-B", 10_000);
    }

    @Test
    void oppositeDirectionsFinishWithoutDeadlockAndPreserveTotal() throws Exception {
        int pairs = 10;
        int tasks = pairs * 2;
        var ready = new CountDownLatch(tasks);
        var start = new CountDownLatch(1);
        var pool = Executors.newFixedThreadPool(tasks);
        try {
            List<Future<?>> futures = new ArrayList<>();
            for (int i = 0; i < pairs; i++) {
                int sequence = i;
                futures.add(pool.submit(() -> invoke(ready, start, "ab-" + sequence,
                    first.getId(), second.getId())));
                futures.add(pool.submit(() -> invoke(ready, start, "ba-" + sequence,
                    second.getId(), first.getId())));
            }
            assertThat(ready.await(10, TimeUnit.SECONDS)).isTrue();
            start.countDown();
            for (Future<?> future : futures) future.get(30, TimeUnit.SECONDS);
            long firstBalance = accounts.findById(first.getId()).orElseThrow().getBalance();
            long secondBalance = accounts.findById(second.getId()).orElseThrow().getBalance();
            assertThat(firstBalance + secondBalance).isEqualTo(20_000);
            assertThat(jdbc.sql("SELECT COALESCE(SUM(signed_amount),0) FROM ledger_entry WHERE entry_type LIKE 'TRANSFER_%'")
                .query(Long.class).single()).isZero();
        } finally {
            start.countDown();
            pool.shutdownNow();
        }
    }

    private Object invoke(
        CountDownLatch ready, CountDownLatch start, String key, long from, long to
    ) {
        try {
            ready.countDown();
            start.await();
            return transfers.transfer(new TransferService.Command("customer-1", key, from, to, 100));
        } catch (InterruptedException interrupted) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(interrupted);
        }
    }
}