package com.easyagents.federation.sql.runtime;

import com.easyagents.federation.sql.adapter.AdapterDialectContext;
import com.easyagents.federation.sql.adapter.AdapterHints;
import com.easyagents.federation.sql.adapter.AdapterSchemaContext;
import com.easyagents.federation.sql.adapter.FederationSqlAdapterProvider;
import com.easyagents.federation.sql.adapter.FederationSqlAdapterRegistry;
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.source.ActiveSourceState;
import com.easyagents.federation.sql.source.FederationDataSourceHandle;
import com.easyagents.federation.sql.source.FederationDataSourceResolver;
import com.easyagents.federation.sql.source.FederationSchemaDefinition;
import com.easyagents.federation.sql.source.FederationSourceDefinition;
import com.easyagents.federation.sql.source.FederationSourceManager;
import com.easyagents.federation.sql.source.FederationSourceState;
import com.easyagents.federation.sql.source.FederationSourceStateProvider;
import com.easyagents.federation.sql.source.FederationSourceView;
import com.easyagents.federation.sql.source.PreparedSourceRuntime;
import com.easyagents.federation.sql.source.SourceApplyOptions;
import com.easyagents.federation.sql.source.SourceApplyResult;
import com.easyagents.federation.sql.source.SourceApplyStatus;
import com.easyagents.federation.sql.source.SourceId;
import com.easyagents.federation.sql.source.SourceProbeResult;
import com.easyagents.federation.sql.source.SourceRemoveResult;
import com.easyagents.federation.sql.source.SourceRuntimeStatus;
import com.easyagents.federation.sql.source.SourceSnapshotResult;
import com.easyagents.federation.sql.source.SourceStateSubscription;
import com.easyagents.federation.sql.source.SourceTombstone;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.SQLException;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.calcite.jdbc.CalciteSchema;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.schema.impl.AbstractSchema;
import org.apache.calcite.sql.SqlDialect;

/**
 * 以 revision、single-flight 初始化和 Runtime lease 管理数据源的默认实现。
 */
public final class DefaultFederationSourceManager implements FederationSourceManager, AutoCloseable {

    private final FederationDataSourceResolver resolver;
    private final FederationSqlAdapterRegistry adapters;
    private final FederationSourceStateProvider stateProvider;
    private final Map<SourceId, SourceSlot> slots = new ConcurrentHashMap<>();
    private final Set<PreparedRuntime> preparedRuntimes = ConcurrentHashMap.newKeySet();
    private final Set<SourceRuntime> retiringRuntimes = ConcurrentHashMap.newKeySet();
    private final List<Consumer<SourceRuntime>> runtimeClosedListeners =
        new CopyOnWriteArrayList<>();
    private final Set<FederationDataSourceHandle> pendingHandleClosures =
        ConcurrentHashMap.newKeySet();
    private final Object sourceCatalogLock = new Object();
    private volatile SourceCatalogSnapshot sourceCatalogSnapshot = new SourceCatalogSnapshot(0, Set.of());
    private final ReentrantReadWriteLock lifecycleLock = new ReentrantReadWriteLock();
    private final Lock operationLock = lifecycleLock.readLock();
    private final Lock closeLock = lifecycleLock.writeLock();
    private final SourceStateSubscription subscription;
    private final AtomicBoolean closed = new AtomicBoolean();
    private final AtomicBoolean subscriptionClosed = new AtomicBoolean();
    private final Object subscriptionCloseLock = new Object();

    /**
     * 创建数据源管理器并恢复共享快照、订阅变更提示。
     *
     * @param resolver DataSource 解析器
     * @param adapters Adapter 注册表
     * @param stateProvider 共享状态 Provider
     */
    public DefaultFederationSourceManager(
        FederationDataSourceResolver resolver,
        FederationSqlAdapterRegistry adapters,
        FederationSourceStateProvider stateProvider
    ) {
        this.resolver = resolver;
        this.adapters = adapters;
        this.stateProvider = stateProvider;
        applySnapshot(stateProvider.loadSnapshot());
        this.subscription = stateProvider.subscribe(this::applySharedState);
    }

    /**
     * 使用 Resolver 探测数据源。
     *
     * @param definition 数据源定义
     * @return 探测结果
     */
    @Override
    public SourceProbeResult probe(FederationSourceDefinition definition) {
        operationLock.lock();
        try {
            ensureOpen();
            return probeLocked(definition, resolver.resolve(definition));
        } finally {
            operationLock.unlock();
        }
    }

    /**
     * 使用临时句柄探测并确定性关闭全部临时资源。
     *
     * @param definition 数据源定义
     * @param temporaryHandle 临时句柄
     * @return 探测结果
     */
    @Override
    public SourceProbeResult probe(
        FederationSourceDefinition definition,
        FederationDataSourceHandle temporaryHandle
    ) {
        operationLock.lock();
        try {
            return probeLocked(definition, temporaryHandle);
        } finally {
            operationLock.unlock();
        }
    }

    private SourceProbeResult probeLocked(
        FederationSourceDefinition definition,
        FederationDataSourceHandle temporaryHandle
    ) {
        SourceRuntime runtime = null;
        boolean buildAttempted = false;
        try {
            ensureOpen();
            buildAttempted = true;
            runtime = buildRuntime(definition, temporaryHandle);
            return new SourceProbeResult(
                definition.sourceId(),
                true,
                temporaryHandle.fingerprint(),
                runtime.compatibility().diagnostic()
            );
        } catch (RuntimeException exception) {
            if (!buildAttempted) {
                closeHandleAndSuppress(temporaryHandle, exception);
            }
            throw exception;
        } finally {
            if (runtime != null) {
                forceCloseRuntime(runtime);
            }
        }
    }

    /**
     * 应用 Definition，并可选择立即预热。
     *
     * @param definition 数据源定义
     * @param options 应用策略
     * @return 应用结果
     */
    @Override
    public SourceApplyResult apply(FederationSourceDefinition definition, SourceApplyOptions options) {
        operationLock.lock();
        try {
            ensureOpen();
            SourceApplyResult result = applyState(ActiveSourceState.of(definition));
            if (options.prewarm() && result.status() != SourceApplyStatus.IGNORED_STALE
                && result.status() != SourceApplyStatus.CONFLICT) {
                ensureReadyInternal(definition.sourceId(), definition.revision());
            }
            return result;
        } finally {
            operationLock.unlock();
        }
    }

    /**
     * 构建与共享 Slot 隔离的 Runtime，供调用方在外部状态 CAS 前验证全部本地资源。
     *
     * @param definition 即将发布的数据源 Definition
     * @return 预构建 Runtime
     */
    @Override
    public PreparedSourceRuntime prepare(FederationSourceDefinition definition) {
        operationLock.lock();
        try {
            ensureOpen();
            FederationDataSourceHandle handle = resolver.resolve(definition);
            SourceRuntime runtime = buildRuntime(definition, handle);
            PreparedRuntime prepared = new PreparedRuntime(runtime);
            preparedRuntimes.add(prepared);
            return prepared;
        } finally {
            operationLock.unlock();
        }
    }

    /**
     * 原子接管并发布由当前管理器创建的预构建 Runtime。
     *
     * @param prepared 预构建 Runtime
     * @return Definition 应用结果
     */
    @Override
    public SourceApplyResult commit(PreparedSourceRuntime prepared) {
        operationLock.lock();
        SourceRuntime candidate = null;
        boolean installed = false;
        try {
            ensureOpen();
            if (!(prepared instanceof PreparedRuntime preparedRuntime)
                || !preparedRuntimes.contains(preparedRuntime)) {
                throw new IllegalArgumentException(
                    "prepared runtime was not created by this source manager or is no longer open"
                );
            }
            candidate = preparedRuntime.take();
            preparedRuntimes.remove(preparedRuntime);
            PreparedCommitOutcome outcome = commitPreparedRuntime(candidate);
            installed = outcome.installed();
            return outcome.result();
        } finally {
            if (candidate != null && !installed) {
                forceCloseRuntime(candidate);
            }
            operationLock.unlock();
        }
    }

    /**
     * 应用共享状态快照并按结果分类计数。
     *
     * @param states 状态集合
     * @return 应用统计
     */
    @Override
    public SourceSnapshotResult applySnapshot(Collection<FederationSourceState> states) {
        operationLock.lock();
        try {
            ensureOpen();
            int applied = 0;
            int idempotent = 0;
            int stale = 0;
            int conflicts = 0;
            if (states == null) {
                return new SourceSnapshotResult(0, 0, 0, 0);
            }
            for (FederationSourceState state : states) {
                SourceApplyResult result = applyState(state);
                switch (result.status()) {
                    case APPLIED -> applied++;
                    case IDEMPOTENT -> idempotent++;
                    case IGNORED_STALE -> stale++;
                    case CONFLICT -> conflicts++;
                }
            }
            return new SourceSnapshotResult(applied, idempotent, stale, conflicts);
        } finally {
            operationLock.unlock();
        }
    }

    /**
     * 应用删除墓碑。
     *
     * @param tombstone 删除墓碑
     * @return 删除结果
     */
    @Override
    public SourceRemoveResult remove(SourceTombstone tombstone) {
        operationLock.lock();
        try {
            ensureOpen();
            return new SourceRemoveResult(applyState(tombstone));
        } finally {
            operationLock.unlock();
        }
    }

    /**
     * 返回节点本地数据源视图。
     *
     * @param sourceId 数据源标识
     * @return 可选视图
     */
    @Override
    public Optional<FederationSourceView> view(SourceId sourceId) {
        operationLock.lock();
        try {
            ensureOpen();
            SourceSlot slot = slots.get(sourceId);
            if (slot == null) {
                return Optional.empty();
            }
            synchronized (slot) {
                return Optional.of(toView(slot));
            }
        } finally {
            operationLock.unlock();
        }
    }

    /**
     * 确保节点本地 Runtime 达到最低 revision；同一 Slot 内初始化天然 single-flight。
     *
     * @param sourceId 数据源标识
     * @param minimumRevision 最低版本
     * @return 就绪视图
     */
    @Override
    public FederationSourceView ensureReady(SourceId sourceId, long minimumRevision) {
        operationLock.lock();
        try {
            ensureOpen();
            return ensureReadyInternal(sourceId, minimumRevision);
        } finally {
            operationLock.unlock();
        }
    }

    private FederationSourceView ensureReadyInternal(SourceId sourceId, long minimumRevision) {
        SourceSlot slot = refreshSlotIfNeeded(sourceId, minimumRevision);
        synchronized (slot) {
            ensureReadyLocked(slot, sourceId, minimumRevision);
            return toView(slot);
        }
    }

    /**
     * 原子确保并获取 Runtime lease，避免 revision 切换与执行获取之间的竞态。
     *
     * @param sourceId 数据源标识
     * @param minimumRevision 最低版本
     * @return Runtime lease
     */
    public SourceRuntime.RuntimeLease acquireRuntime(SourceId sourceId, long minimumRevision) {
        operationLock.lock();
        try {
            ensureOpen();
            SourceSlot slot = refreshSlotIfNeeded(sourceId, minimumRevision);
            synchronized (slot) {
                ensureReadyLocked(slot, sourceId, minimumRevision);
                return slot.current.acquire();
            }
        } finally {
            operationLock.unlock();
        }
    }

    /**
     * 原子返回注册表代次与 SourceId 集合，供计划缓存和编译共享同一语义快照。
     *
     * @return 注册表快照
     */
    SourceCatalogSnapshot catalogSnapshot() {
        operationLock.lock();
        try {
            ensureOpen();
            return sourceCatalogSnapshot;
        } finally {
            operationLock.unlock();
        }
    }

    private SourceSlot refreshSlotIfNeeded(SourceId sourceId, long minimumRevision) {
        SourceSlot slot = slots.get(sourceId);
        if (slot == null || slot.desired.revision() < minimumRevision) {
            stateProvider.find(sourceId).ifPresent(this::applySharedState);
            slot = slots.get(sourceId);
        }
        if (slot == null) {
            throw new FederationSqlException(
                FederationSqlErrorCode.SOURCE_NOT_FOUND,
                "source is not defined: " + sourceId
            );
        }
        return slot;
    }

    private void ensureReadyLocked(SourceSlot slot, SourceId sourceId, long minimumRevision) {
        if (slot.desired instanceof SourceTombstone) {
            throw new FederationSqlException(
                FederationSqlErrorCode.SOURCE_REMOVED,
                "source has been removed: " + sourceId
            );
        }
        ActiveSourceState active = (ActiveSourceState) slot.desired;
        if (active.revision() < minimumRevision) {
            throw new FederationSqlException(
                FederationSqlErrorCode.SOURCE_REVISION_NOT_READY,
                "source revision " + active.revision() + " is lower than required " + minimumRevision
            );
        }
        if (slot.current != null && slot.current.definition().revision() >= minimumRevision
            && slot.current.definition().revision() == active.revision()) {
            return;
        }

        SourceRuntime candidate;
        try {
            FederationDataSourceHandle handle = resolver.resolve(active.definition());
            candidate = buildRuntime(active.definition(), handle);
        } catch (FederationSqlException exception) {
            slot.lastFailure = exception.getMessage();
            if (slot.current != null && slot.current.definition().revision() >= minimumRevision) {
                return;
            }
            throw exception;
        } catch (RuntimeException exception) {
            slot.lastFailure = exception.getMessage();
            if (slot.current != null && slot.current.definition().revision() >= minimumRevision) {
                return;
            }
            throw new FederationSqlException(
                FederationSqlErrorCode.SOURCE_INITIALIZATION_FAILED,
                "failed to initialize source " + sourceId,
                exception
            );
        }

        SourceRuntime previous = slot.current;
        slot.current = candidate;
        slot.lastFailure = null;
        if (previous != null) {
            retireRuntime(previous);
        }
    }

    private SourceRuntime buildRuntime(
        FederationSourceDefinition definition,
        FederationDataSourceHandle handle
    ) {
        try {
            FederationSqlAdapterProvider adapter = adapters.require(definition.adapterId());
            try (Connection connection = handle.dataSource().getConnection()) {
                DatabaseMetaData metadata = connection.getMetaData();
                AdapterHints hints = new AdapterHints(definition.adapterOptions());
                if (!adapter.supports(metadata, hints)) {
                    throw new FederationSqlException(
                        FederationSqlErrorCode.ADAPTER_UNSUPPORTED,
                        "adapter " + adapter.adapterId() + " does not support database "
                            + metadata.getDatabaseProductName()
                    );
                }
                SqlDialect dialect = adapter.createDialect(new AdapterDialectContext(metadata, definition));
                // 动态 JDBC 元数据必须在冷编译时可见；有界计划缓存负责热查询性能。
                SchemaPlus root = CalciteSchema.createRootSchema(true, false).plus();
                SchemaPlus sourceSchema = root.add(definition.sourceId().value(), new AbstractSchema());
                SchemaPlus onlySchema = null;
                for (FederationSchemaDefinition schemaDefinition : definition.schemas()) {
                    Schema schema = adapter.createSchema(new AdapterSchemaContext(
                        sourceSchema,
                        definition,
                        schemaDefinition,
                        handle,
                        dialect
                    ));
                    SchemaPlus added = sourceSchema.add(schemaDefinition.logicalName(), schema);
                    onlySchema = definition.schemas().size() == 1 ? added : null;
                }
                SchemaPlus defaultSchema = onlySchema == null ? sourceSchema : onlySchema;
                return new SourceRuntime(
                    definition,
                    handle,
                    adapter,
                    dialect,
                    root,
                    defaultSchema,
                    adapter.compatibility(metadata, hints),
                    this::runtimeClosed
                );
            }
        } catch (SQLException exception) {
            FederationSqlException failure = new FederationSqlException(
                FederationSqlErrorCode.SOURCE_INITIALIZATION_FAILED,
                "failed to inspect source " + definition.sourceId(),
                exception
            );
            closeHandleAndSuppress(handle, failure);
            throw failure;
        } catch (RuntimeException exception) {
            closeHandleAndSuppress(handle, exception);
            throw exception;
        }
    }

    private void closeHandleAndSuppress(
        FederationDataSourceHandle handle,
        RuntimeException original
    ) {
        try {
            handle.close();
        } catch (RuntimeException closeException) {
            pendingHandleClosures.add(handle);
            original.addSuppressed(closeException);
        }
    }

    private void applySharedState(FederationSourceState state) {
        operationLock.lock();
        try {
            if (!closed.get()) {
                applyState(state);
            }
        } finally {
            operationLock.unlock();
        }
    }

    private SourceApplyResult applyState(FederationSourceState state) {
        synchronized (sourceCatalogLock) {
            SourceSlot newSlot = new SourceSlot(state);
            SourceSlot existing = slots.putIfAbsent(state.sourceId(), newSlot);
            if (existing == null) {
                if (state instanceof ActiveSourceState) {
                    publishSourceCatalogSnapshot();
                }
                return result(state, state.revision(), SourceApplyStatus.APPLIED);
            }
            SourceSlot slot = existing;
            synchronized (slot) {
                FederationSourceState current = slot.desired;
                if (state.revision() < current.revision()) {
                    return result(state, current.revision(), SourceApplyStatus.IGNORED_STALE);
                }
                if (state.revision() == current.revision()) {
                    SourceApplyStatus status = state.checksum().equals(current.checksum())
                        ? SourceApplyStatus.IDEMPOTENT
                        : SourceApplyStatus.CONFLICT;
                    return result(state, current.revision(), status);
                }
                slot.desired = state;
                if ((current instanceof ActiveSourceState) != (state instanceof ActiveSourceState)) {
                    publishSourceCatalogSnapshot();
                }
                if (state instanceof SourceTombstone && slot.current != null) {
                    retireRuntime(slot.current);
                    slot.current = null;
                }
                return result(state, state.revision(), SourceApplyStatus.APPLIED);
            }
        }
    }

    private PreparedCommitOutcome commitPreparedRuntime(SourceRuntime candidate) {
        ActiveSourceState state = ActiveSourceState.of(candidate.definition());
        synchronized (sourceCatalogLock) {
            SourceSlot newSlot = new SourceSlot(state);
            newSlot.current = candidate;
            SourceSlot existing = slots.putIfAbsent(state.sourceId(), newSlot);
            if (existing == null) {
                publishSourceCatalogSnapshot();
                return new PreparedCommitOutcome(
                    result(state, state.revision(), SourceApplyStatus.APPLIED), true
                );
            }
            SourceSlot slot = existing;
            synchronized (slot) {
                FederationSourceState current = slot.desired;
                if (state.revision() < current.revision()) {
                    return new PreparedCommitOutcome(
                        result(state, current.revision(), SourceApplyStatus.IGNORED_STALE), false
                    );
                }
                SourceApplyStatus status;
                if (state.revision() == current.revision()) {
                    if (!state.checksum().equals(current.checksum())) {
                        return new PreparedCommitOutcome(
                            result(state, current.revision(), SourceApplyStatus.CONFLICT), false
                        );
                    }
                    status = SourceApplyStatus.IDEMPOTENT;
                } else {
                    status = SourceApplyStatus.APPLIED;
                    slot.desired = state;
                    if (current instanceof SourceTombstone) {
                        publishSourceCatalogSnapshot();
                    }
                }
                if (slot.current != null
                    && slot.current.definition().revision() == state.revision()
                    && slot.current.sourceChecksum().equals(state.checksum())) {
                    return new PreparedCommitOutcome(
                        result(state, state.revision(), status), false
                    );
                }
                SourceRuntime previous = slot.current;
                slot.current = candidate;
                slot.lastFailure = null;
                if (previous != null) {
                    try {
                        retireRuntime(previous);
                    } catch (RuntimeException exception) {
                        // 新 Runtime 已原子接管；记录旧 Handle 关闭错误且不回滚可用的新版本。
                        slot.lastFailure = "previous runtime retirement failed: " + exception.getMessage();
                    }
                }
                return new PreparedCommitOutcome(
                    result(state, state.revision(), status), true
                );
            }
        }
    }

    private Set<SourceId> activeSourceIds() {
        Set<SourceId> sourceIds = new HashSet<>();
        slots.forEach((sourceId, slot) -> {
            if (slot.desired instanceof ActiveSourceState) {
                sourceIds.add(sourceId);
            }
        });
        return Set.copyOf(sourceIds);
    }

    private void publishSourceCatalogSnapshot() {
        sourceCatalogSnapshot = new SourceCatalogSnapshot(
            sourceCatalogSnapshot.generation() + 1,
            activeSourceIds()
        );
    }

    private static SourceApplyResult result(
        FederationSourceState requested,
        long effectiveRevision,
        SourceApplyStatus status
    ) {
        return new SourceApplyResult(
            requested.sourceId(),
            requested.revision(),
            effectiveRevision,
            status
        );
    }

    private static FederationSourceView toView(SourceSlot slot) {
        SourceRuntimeStatus status;
        if (slot.desired instanceof SourceTombstone) {
            status = SourceRuntimeStatus.REMOVED;
        } else if (slot.current != null) {
            status = SourceRuntimeStatus.READY;
        } else if (slot.lastFailure != null) {
            status = SourceRuntimeStatus.FAILED;
        } else {
            status = SourceRuntimeStatus.DEFINED;
        }
        return new FederationSourceView(
            slot.desired.sourceId(),
            slot.desired.revision(),
            slot.current == null ? -1 : slot.current.definition().revision(),
            status,
            slot.desired.checksum(),
            slot.lastFailure
        );
    }

    private void ensureOpen() {
        if (closed.get()) {
            throw new FederationSqlException(FederationSqlErrorCode.ENGINE_CLOSED, "source manager is closed");
        }
    }

    /**
     * 登记旧 Runtime 并按租约排空语义关闭。
     *
     * @param runtime 待退役 Runtime
     */
    private void retireRuntime(SourceRuntime runtime) {
        retiringRuntimes.add(runtime);
        runtime.retire();
    }

    /**
     * 增加节点本地 Runtime 成功关闭后的监听器。
     *
     * @param listener 关闭监听器
     */
    void addRuntimeClosedListener(Consumer<SourceRuntime> listener) {
        runtimeClosedListeners.add(java.util.Objects.requireNonNull(listener, "listener"));
    }

    private void runtimeClosed(SourceRuntime runtime) {
        retiringRuntimes.remove(runtime);
        RuntimeException failure = null;
        for (Consumer<SourceRuntime> listener : runtimeClosedListeners) {
            try {
                listener.accept(runtime);
            } catch (RuntimeException exception) {
                if (failure == null) {
                    failure = exception;
                } else {
                    failure.addSuppressed(exception);
                }
            }
        }
        if (failure != null) {
            throw failure;
        }
    }

    /**
     * 登记候选 Runtime 并立即关闭其 Handle。
     *
     * @param runtime 待关闭 Runtime
     */
    private void forceCloseRuntime(SourceRuntime runtime) {
        retiringRuntimes.add(runtime);
        runtime.forceClose();
    }

    /**
     * 关闭订阅并让所有 Runtime 进入排空；无活跃 lease 的 Handle 会立即关闭。
     */
    @Override
    public void close() {
        RuntimeException failure = null;
        closeLock.lock();
        try {
            // write lock 排除了并发 build/commit，快照包含进入关闭前的全部失败资源。
            Set<SourceRuntime> retryRuntimes = Set.copyOf(retiringRuntimes);
            Set<FederationDataSourceHandle> retryHandles =
                Set.copyOf(pendingHandleClosures);
            if (closed.compareAndSet(false, true)) {
                for (SourceSlot slot : slots.values()) {
                    synchronized (slot) {
                        if (slot.current != null) {
                            try {
                                retireRuntime(slot.current);
                            } catch (RuntimeException exception) {
                                failure = append(failure, exception);
                            }
                            slot.current = null;
                        }
                    }
                }
                for (PreparedRuntime prepared : Set.copyOf(preparedRuntimes)) {
                    try {
                        prepared.close();
                    } catch (RuntimeException exception) {
                        failure = append(failure, exception);
                    }
                }
            }
            // 上一次释放失败的资源在后续 close 调用中重试，成功后由回调移出集合。
            for (SourceRuntime runtime : retryRuntimes) {
                try {
                    runtime.retire();
                } catch (RuntimeException exception) {
                    failure = append(failure, exception);
                }
            }
            for (FederationDataSourceHandle handle : retryHandles) {
                try {
                    handle.close();
                    pendingHandleClosures.remove(handle);
                } catch (RuntimeException exception) {
                    failure = append(failure, exception);
                }
            }
        } finally {
            closeLock.unlock();
        }
        synchronized (subscriptionCloseLock) {
            try {
                if (!subscriptionClosed.get()) {
                    subscription.close();
                    subscriptionClosed.set(true);
                }
            } catch (RuntimeException exception) {
                failure = append(failure, exception);
            }
        }
        if (failure != null) {
            throw failure;
        }
    }

    private static RuntimeException append(RuntimeException failure, RuntimeException next) {
        if (failure == null) {
            return next;
        }
        failure.addSuppressed(next);
        return failure;
    }

    private static final class SourceSlot {

        private FederationSourceState desired;
        private SourceRuntime current;
        private String lastFailure;

        private SourceSlot(FederationSourceState desired) {
            this.desired = desired;
        }
    }

    private record PreparedCommitOutcome(
        SourceApplyResult result,
        boolean installed
    ) {
    }

    private final class PreparedRuntime implements PreparedSourceRuntime {

        private SourceRuntime runtime;

        private PreparedRuntime(SourceRuntime runtime) {
            this.runtime = runtime;
        }

        @Override
        public synchronized FederationSourceDefinition definition() {
            if (runtime == null) {
                throw new IllegalStateException("prepared runtime is no longer open");
            }
            return runtime.definition();
        }

        private synchronized SourceRuntime take() {
            if (runtime == null) {
                throw new IllegalStateException("prepared runtime is no longer open");
            }
            SourceRuntime claimed = runtime;
            runtime = null;
            return claimed;
        }

        @Override
        public void close() {
            SourceRuntime discarded;
            synchronized (this) {
                discarded = runtime;
                runtime = null;
            }
            preparedRuntimes.remove(this);
            if (discarded != null) {
                forceCloseRuntime(discarded);
            }
        }
    }
}
