데이터 접근 기술 전환

package board.migration;

import java.time.Duration;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;

import board.application.PostPage;
import board.application.PostQuery;
import board.application.PostSearch;
import board.application.PostSnapshot;
import board.application.PostSummary;

public final class VerifyingPostQuery implements PostQuery {
    private static final Duration MAXIMUM_TIMEOUT = Duration.ofSeconds(2);

    private final PostQuery primary;
    private final PostQuery candidate;
    private final ThreadPoolExecutor candidateExecutor;
    private final ScheduledExecutorService timeoutScheduler;
    private final Duration candidateTimeout;
    private final Sampler sampler;
    private final DifferenceSink differences;

    public VerifyingPostQuery(
            PostQuery primary,
            PostQuery candidate,
            ThreadPoolExecutor candidateExecutor,
            ScheduledExecutorService timeoutScheduler,
            Duration candidateTimeout,
            Sampler sampler,
            DifferenceSink differences
    ) {
        this.primary = Objects.requireNonNull(primary, "primary");
        this.candidate = Objects.requireNonNull(candidate, "candidate");
        this.candidateExecutor = Objects.requireNonNull(
                candidateExecutor, "candidateExecutor");
        this.timeoutScheduler = Objects.requireNonNull(
                timeoutScheduler, "timeoutScheduler");
        this.candidateTimeout = requireTimeout(candidateTimeout);
        this.sampler = Objects.requireNonNull(sampler, "sampler");
        this.differences = Objects.requireNonNull(differences, "differences");
        long queueCapacity = (long) candidateExecutor.getQueue().size()
                + candidateExecutor.getQueue().remainingCapacity();
        if (queueCapacity >= Integer.MAX_VALUE
                || !(candidateExecutor.getRejectedExecutionHandler()
                        instanceof ThreadPoolExecutor.AbortPolicy)) {
            throw new IllegalArgumentException(
                    "candidateExecutor must be bounded and reject with AbortPolicy");
        }
    }

    @Override
    public Optional<PostSnapshot> find(long postId, long memberId) {
        Optional<PostSnapshot> response = primary.find(postId, memberId);
        compare(
                "find",
                response,
                () -> candidate.find(postId, memberId));
        return response;
    }

    @Override
    public Optional<PostSnapshot> findByIdempotencyKey(
            long memberId,
            String clientRequestId
    ) {
        Optional<PostSnapshot> response = primary.findByIdempotencyKey(
                memberId, clientRequestId);
        compare(
                "findByIdempotencyKey",
                response,
                () -> candidate.findByIdempotencyKey(
                        memberId, clientRequestId));
        return response;
    }

    @Override
    public PostPage findPage(PostSearch search) {
        PostPage response = primary.findPage(search);
        compare("findPage", response, () -> candidate.findPage(search));
        return response;
    }

    @Override
    public PostSummary summary(long memberId) {
        PostSummary response = primary.summary(memberId);
        compare("summary", response, () -> candidate.summary(memberId));
        return response;
    }

    private <T> void compare(
            String operation,
            T primaryResult,
            Supplier<T> candidateRead
    ) {
        if (!selected()) {
            return;
        }
        ComparisonAttempt attempt = new ComparisonAttempt(operation);
        try {
            ScheduledFuture<?> deadline = timeoutScheduler.schedule(
                    () -> attempt.complete(Outcome.TIMEOUT, null),
                    candidateTimeout.toNanos(),
                    TimeUnit.NANOSECONDS);
            attempt.arm(deadline);
            candidateExecutor.execute(() -> {
                if (attempt.isComplete()) {
                    return;
                }
                try {
                    T candidateResult = candidateRead.get();
                    Outcome outcome = Objects.equals(
                            primaryResult, candidateResult)
                            ? Outcome.MATCH
                            : Outcome.MISMATCH;
                    attempt.complete(outcome, null);
                } catch (RuntimeException candidateFailure) {
                    attempt.complete(Outcome.FAILURE, candidateFailure);
                }
            });
        } catch (RejectedExecutionException rejected) {
            attempt.complete(Outcome.REJECTED, rejected);
        }
    }

    private boolean selected() {
        try {
            return sampler.selected();
        } catch (RuntimeException samplerFailure) {
            return false;
        }
    }

    private static Duration requireTimeout(Duration timeout) {
        Objects.requireNonNull(timeout, "candidateTimeout");
        if (timeout.isZero()
                || timeout.isNegative()
                || timeout.compareTo(MAXIMUM_TIMEOUT) > 0) {
            throw new IllegalArgumentException(
                    "candidateTimeout must be in (0, 2 seconds]");
        }
        return timeout;
    }

    private final class ComparisonAttempt {
        private final String operation;
        private final AtomicBoolean completed = new AtomicBoolean();
        private final AtomicReference<ScheduledFuture<?>> deadline =
                new AtomicReference<>();

        private ComparisonAttempt(String operation) {
            this.operation = operation;
        }

        private void arm(ScheduledFuture<?> scheduled) {
            if (!deadline.compareAndSet(null, scheduled)) {
                scheduled.cancel(false);
                throw new IllegalStateException("deadline already armed");
            }
            if (completed.get()) {
                scheduled.cancel(false);
            }
        }

        private void complete(Outcome outcome, RuntimeException failure) {
            if (!completed.compareAndSet(false, true)) {
                return;
            }
            ScheduledFuture<?> scheduled = deadline.get();
            if (scheduled != null) {
                scheduled.cancel(false);
            }
            String failureType = failure == null
                    ? null
                    : failure.getClass().getName();
            try {
                differences.record(new ComparisonEvent(
                        operation, outcome, failureType));
            } catch (RuntimeException sinkFailure) {
                // Diagnostics cannot replace or fail the primary response.
            }
        }

        private boolean isComplete() {
            return completed.get();
        }
    }

    @FunctionalInterface
    public interface Sampler {
        boolean selected();
    }

    @FunctionalInterface
    public interface DifferenceSink {
        void record(ComparisonEvent event);
    }

    public enum Outcome {
        MATCH,
        MISMATCH,
        FAILURE,
        TIMEOUT,
        REJECTED
    }

    public record ComparisonEvent(
            String operation,
            Outcome outcome,
            String failureType
    ) {
        public ComparisonEvent {
            if (operation == null || operation.isBlank() || outcome == null) {
                throw new IllegalArgumentException("invalid comparison event");
            }
            boolean failureOutcome = outcome == Outcome.FAILURE
                    || outcome == Outcome.REJECTED;
            if (failureOutcome
                    != (failureType != null && !failureType.isBlank())) {
                throw new IllegalArgumentException(
                        "failure type must match the outcome");
            }
        }
    }
}

package board.migration;

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

import java.time.Duration;
import java.time.Instant;
import java.time.LocalDate;
import java.util.Optional;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;

import board.application.PostPage;
import board.application.PostQuery;
import board.application.PostSearch;
import board.application.PostSnapshot;
import board.application.PostSummary;
import board.migration.VerifyingPostQuery.ComparisonEvent;
import board.migration.VerifyingPostQuery.Outcome;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;

class VerifyingPostQueryTest {
    private static final PostSnapshot PRIMARY = new PostSnapshot(
            1L,
            41L,
            "Persistence",
            "primary content",
            LocalDate.parse("2026-07-14"),
            "migration-01",
            Instant.parse("2026-07-14T09:00:00Z"),
            0L);

    @Test
    void mismatch_returns_primary_and_emits_one_redacted_event()
            throws Exception {
        PostSnapshot different = new PostSnapshot(
                PRIMARY.id(),
                PRIMARY.memberId(),
                PRIMARY.title(),
                "candidate content",
                PRIMARY.publishedOn(),
                PRIMARY.clientRequestId(),
                PRIMARY.createdAt(),
                PRIMARY.version());
        FindOnlyQuery candidate = new FindOnlyQuery(
                (id, memberId) -> Optional.of(different));
        RecordingSink sink = new RecordingSink();

        try (ThreadPoolExecutor executor = boundedExecutor(8);
                ScheduledExecutorService scheduler =
                        Executors.newSingleThreadScheduledExecutor()) {
            VerifyingPostQuery query = query(
                    candidate, executor, scheduler, Duration.ofSeconds(1),
                    true, sink);

            Optional<PostSnapshot> response = query.find(1L, 41L);

            assertThat(response.orElseThrow()).isSameAs(PRIMARY);
            assertEvent(sink.await(), Outcome.MISMATCH, null);
            assertThat(sink.poll()).isNull();
            assertThat(candidate.calls()).isEqualTo(1);
        }
    }

    @Test
    void candidate_failure_returns_primary_and_emits_one_failure_event()
            throws Exception {
        FindOnlyQuery candidate = new FindOnlyQuery((id, memberId) -> {
            throw new IllegalStateException("candidate unavailable");
        });
        RecordingSink sink = new RecordingSink();

        try (ThreadPoolExecutor executor = boundedExecutor(8);
                ScheduledExecutorService scheduler =
                        Executors.newSingleThreadScheduledExecutor()) {
            VerifyingPostQuery query = query(
                    candidate, executor, scheduler, Duration.ofSeconds(1),
                    true, sink);

            Optional<PostSnapshot> response = query.find(1L, 41L);

            assertThat(response.orElseThrow()).isSameAs(PRIMARY);
            assertEvent(
                    sink.await(),
                    Outcome.FAILURE,
                    IllegalStateException.class.getName());
            assertThat(sink.poll()).isNull();
        }
    }

    @Test
    @Timeout(4)
    void timeout_does_not_wait_and_late_candidate_cannot_emit_twice()
            throws Exception {
        CountDownLatch started = new CountDownLatch(1);
        CountDownLatch release = new CountDownLatch(1);
        CountDownLatch finished = new CountDownLatch(1);
        FindOnlyQuery candidate = new FindOnlyQuery((id, memberId) -> {
            started.countDown();
            try {
                await(release);
                return Optional.of(PRIMARY);
            } finally {
                finished.countDown();
            }
        });
        RecordingSink sink = new RecordingSink();

        try (ThreadPoolExecutor executor = boundedExecutor(8);
                ScheduledExecutorService scheduler =
                        Executors.newSingleThreadScheduledExecutor()) {
            executor.prestartCoreThread();
            VerifyingPostQuery query = query(
                    candidate, executor, scheduler, Duration.ofMillis(100),
                    true, sink);

            Optional<PostSnapshot> response = query.find(1L, 41L);

            assertThat(response.orElseThrow()).isSameAs(PRIMARY);
            assertThat(started.await(1, TimeUnit.SECONDS)).isTrue();
            assertEvent(sink.await(), Outcome.TIMEOUT, null);
            release.countDown();
            assertThat(finished.await(1, TimeUnit.SECONDS)).isTrue();
            assertThat(sink.poll()).isNull();
        } finally {
            release.countDown();
        }
    }

    @Test
    void saturated_executor_returns_primary_and_emits_one_rejection()
            throws Exception {
        CountDownLatch workerStarted = new CountDownLatch(1);
        CountDownLatch releaseWorker = new CountDownLatch(1);
        FindOnlyQuery candidate = new FindOnlyQuery(
                (id, memberId) -> Optional.of(PRIMARY));
        RecordingSink sink = new RecordingSink();

        try (ThreadPoolExecutor executor = boundedExecutor(1);
                ScheduledExecutorService scheduler =
                        Executors.newSingleThreadScheduledExecutor()) {
            try {
                executor.execute(() -> {
                    workerStarted.countDown();
                    await(releaseWorker);
                });
                assertThat(workerStarted.await(1, TimeUnit.SECONDS)).isTrue();
                executor.execute(() -> { });
                VerifyingPostQuery query = query(
                        candidate, executor, scheduler, Duration.ofSeconds(1),
                        true, sink);

                Optional<PostSnapshot> response = query.find(1L, 41L);

                assertThat(response.orElseThrow()).isSameAs(PRIMARY);
                assertEvent(
                        sink.await(),
                        Outcome.REJECTED,
                        java.util.concurrent.RejectedExecutionException.class
                                .getName());
                assertThat(candidate.calls()).isZero();
                assertThat(sink.poll()).isNull();
            } finally {
                releaseWorker.countDown();
            }
        }
    }

    @Test
    void unsampled_read_never_submits_candidate_work() throws Exception {
        FindOnlyQuery candidate = new FindOnlyQuery(
                (id, memberId) -> Optional.of(PRIMARY));
        RecordingSink sink = new RecordingSink();

        try (ThreadPoolExecutor executor = boundedExecutor(8);
                ScheduledExecutorService scheduler =
                        Executors.newSingleThreadScheduledExecutor()) {
            VerifyingPostQuery query = query(
                    candidate, executor, scheduler, Duration.ofSeconds(1),
                    false, sink);

            Optional<PostSnapshot> response = query.find(1L, 41L);

            assertThat(response.orElseThrow()).isSameAs(PRIMARY);
            assertThat(candidate.calls()).isZero();
            assertThat(sink.poll()).isNull();
        }
    }

    @Test
    void cutover_gate_fails_closed_for_coverage_mismatch_latency_or_failure() {
        MigrationGate gate = new MigrationGate(
                0.25,
                0.01,
                Duration.ofMillis(200),
                1_000,
                Duration.ofMillis(50));

        assertThat(gate.permitsCutover(new MigrationGate.Observation(
                1_000, 0, 0, 0, 0, Duration.ofMillis(40)))).isTrue();
        assertThat(gate.permitsCutover(new MigrationGate.Observation(
                999, 0, 0, 0, 0, Duration.ofMillis(40)))).isFalse();
        assertThat(gate.permitsCutover(new MigrationGate.Observation(
                1_000, 11, 0, 0, 0, Duration.ofMillis(40)))).isFalse();
        assertThat(gate.permitsCutover(new MigrationGate.Observation(
                1_000, 0, 0, 0, 0, Duration.ofMillis(51)))).isFalse();
        assertThat(gate.permitsCutover(new MigrationGate.Observation(
                1_000, 0, 1, 0, 0, Duration.ofMillis(40)))).isFalse();
        assertThat(gate.permitsCutover(new MigrationGate.Observation(
                1_000, 0, 0, 1, 0, Duration.ofMillis(40)))).isFalse();
        assertThat(gate.permitsCutover(new MigrationGate.Observation(
                1_000, 0, 0, 0, 1, Duration.ofMillis(40)))).isFalse();
    }

    private static VerifyingPostQuery query(
            FindOnlyQuery candidate,
            ThreadPoolExecutor executor,
            ScheduledExecutorService scheduler,
            Duration timeout,
            boolean sampled,
            RecordingSink sink
    ) {
        return new VerifyingPostQuery(
                new FindOnlyQuery((id, memberId) -> Optional.of(PRIMARY)),
                candidate,
                executor,
                scheduler,
                timeout,
                () -> sampled,
                sink);
    }

    private static ThreadPoolExecutor boundedExecutor(int queueCapacity) {
        return new ThreadPoolExecutor(
                1,
                1,
                0L,
                TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(queueCapacity),
                new ThreadPoolExecutor.AbortPolicy());
    }

    private static void assertEvent(
            ComparisonEvent event,
            Outcome outcome,
            String failureType
    ) {
        assertThat(event.operation()).isEqualTo("find");
        assertThat(event.outcome()).isEqualTo(outcome);
        assertThat(event.failureType()).isEqualTo(failureType);
    }

    private static void await(CountDownLatch latch) {
        try {
            latch.await();
        } catch (InterruptedException interrupted) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("interrupted", interrupted);
        }
    }

    private static final class FindOnlyQuery implements PostQuery {
        private final BiFunction<Long, Long, Optional<PostSnapshot>> behavior;
        private final AtomicInteger calls = new AtomicInteger();

        private FindOnlyQuery(
                BiFunction<Long, Long, Optional<PostSnapshot>> behavior
        ) {
            this.behavior = behavior;
        }

        @Override
        public Optional<PostSnapshot> find(long postId, long memberId) {
            calls.incrementAndGet();
            return behavior.apply(postId, memberId);
        }

        @Override
        public Optional<PostSnapshot> findByIdempotencyKey(
                long memberId,
                String clientRequestId
        ) {
            throw new AssertionError("not used by this test");
        }

        @Override
        public PostPage findPage(PostSearch search) {
            throw new AssertionError("not used by this test");
        }

        @Override
        public PostSummary summary(long memberId) {
            throw new AssertionError("not used by this test");
        }

        private int calls() {
            return calls.get();
        }
    }

    private static final class RecordingSink
            implements VerifyingPostQuery.DifferenceSink {
        private final BlockingQueue<ComparisonEvent> events =
                new LinkedBlockingQueue<>();

        @Override
        public void record(ComparisonEvent event) {
            events.add(event);
        }

        private ComparisonEvent await() throws InterruptedException {
            ComparisonEvent event = events.poll(1, TimeUnit.SECONDS);
            assertThat(event).isNotNull();
            return event;
        }

        private ComparisonEvent poll() throws InterruptedException {
            return events.poll(100, TimeUnit.MILLISECONDS);
        }
    }
}

package board.migration;

import java.time.Duration;
import java.util.Objects;

public record MigrationGate(
        double shadowRatio,
        double maximumMismatchRatio,
        Duration candidateTimeout,
        long minimumSamples,
        Duration maximumP95Latency
) {
    public MigrationGate {
        Objects.requireNonNull(candidateTimeout, "candidateTimeout");
        Objects.requireNonNull(maximumP95Latency, "maximumP95Latency");
        if (!Double.isFinite(shadowRatio)
                || shadowRatio < 0.0
                || shadowRatio > 1.0
                || !Double.isFinite(maximumMismatchRatio)
                || maximumMismatchRatio < 0.0
                || maximumMismatchRatio > 1.0
                || candidateTimeout.isZero()
                || candidateTimeout.isNegative()
                || candidateTimeout.compareTo(Duration.ofSeconds(2)) > 0
                || minimumSamples < 1
                || maximumP95Latency.isZero()
                || maximumP95Latency.isNegative()
                || maximumP95Latency.compareTo(candidateTimeout) > 0) {
            throw new IllegalArgumentException("invalid migration gate");
        }
    }

    public boolean permitsCutover(Observation observation) {
        Objects.requireNonNull(observation, "observation");
        if (shadowRatio == 0.0
                || observation.samples() < minimumSamples
                || observation.failures() > 0
                || observation.timeouts() > 0
                || observation.rejections() > 0
                || observation.p95Latency().compareTo(maximumP95Latency) > 0) {
            return false;
        }
        double mismatchRatio = (double) observation.mismatches()
                / (double) observation.samples();
        return mismatchRatio <= maximumMismatchRatio;
    }

    public record Observation(
            long samples,
            long mismatches,
            long failures,
            long timeouts,
            long rejections,
            Duration p95Latency
    ) {
        public Observation {
            Objects.requireNonNull(p95Latency, "p95Latency");
            boolean invalid = samples < 0
                    || mismatches < 0
                    || failures < 0
                    || timeouts < 0
                    || rejections < 0
                    || p95Latency.isNegative();
            long remaining = samples;
            if (!invalid && mismatches <= remaining) {
                remaining -= mismatches;
            } else if (!invalid) {
                invalid = true;
            }
            if (!invalid && failures <= remaining) {
                remaining -= failures;
            } else if (!invalid) {
                invalid = true;
            }
            if (!invalid && timeouts <= remaining) {
                remaining -= timeouts;
            } else if (!invalid) {
                invalid = true;
            }
            if (!invalid && rejections > remaining) {
                invalid = true;
            }
            if (invalid) {
                throw new IllegalArgumentException("invalid observation");
            }
        }
    }
}