flowchart TB
    subgraph "客户端层"
        Client[REST Client<br/>_bulk / _doc APIs]
    end

    subgraph "请求处理层"
        TransportAction[TransportBulkAction<br/>TransportIndexAction]
        BulkProcessor[BulkProcessor<br/>批量处理器]
    end

    subgraph "分片管理层"
        IndexShard[IndexShard<br/>分片管理]
        ReplicationOp[ReplicationOperation<br/>复制协调]
    end

    subgraph "存储引擎层 (Engine)"
        Engine[Engine<br/>抽象接口]
        InternalEngine[InternalEngine<br/>读写实现]
        ReadOnlyEngine[ReadOnlyEngine<br/>只读实现]
    end

    subgraph "核心组件层"
        VersionMap[LiveVersionMap<br/>版本映射]
        Translog[Translog<br/>事务日志]
        LocalCheckpoint[LocalCheckpointTracker<br/>检查点跟踪]
        Throttle[IndexThrottle<br/>索引限流]
    end

    subgraph "Lucene 层"
        IndexWriter[IndexWriter<br/>写入器]
        DirectoryReader[DirectoryReader<br/>读取器]
        ReaderManager[ReaderManager<br/>Reader管理]
        SegmentInfos[SegmentInfos<br/>段元数据]
    end

    subgraph "存储层"
        Directory[FSDirectory<br/>文件目录]
        TranslogFiles[Translog Files<br/>translog-N.tlog]
        SegmentFiles[Segment Files<br/>segments_N]
    end

    Client --> TransportAction
    TransportAction --> BulkProcessor
    BulkProcessor --> IndexShard
    IndexShard --> ReplicationOp
    ReplicationOp --> |Primary| IndexShard
    ReplicationOp --> |Replica| IndexShard

    IndexShard --> Engine
    Engine --> InternalEngine
    Engine --> ReadOnlyEngine

    InternalEngine --> VersionMap
    InternalEngine --> Translog
    InternalEngine --> LocalCheckpoint
    InternalEngine --> Throttle
    InternalEngine --> IndexWriter
    InternalEngine --> ReaderManager

    IndexWriter --> SegmentInfos
    IndexWriter --> Directory
    ReaderManager --> DirectoryReader
    DirectoryReader --> Directory

    Translog --> TranslogFiles
    Directory --> SegmentFiles

    style InternalEngine fill:#e1f5ff
    style VersionMap fill:#e1ffe1
    style Translog fill:#ffe1e1
    style IndexWriter fill:#fff4e1

POST /index/_doc/1 → TransportIndexAction
POST /_bulk → TransportBulkAction → BulkProcessor

flowchart TB
    Start[客户端请求<br/>POST /index/_doc/1] --> TransportAction[TransportIndexAction]
    TransportAction --> |路由到分片| IndexShard[IndexShard]

    IndexShard --> PrepareCheck{是Primary?}
    PrepareCheck --> |Yes| PrimaryPath[applyIndexOperationOnPrimary]
    PrepareCheck --> |No| ReplicaPath[applyIndexOperationOnReplica]

    PrimaryPath --> PrepareIndex[prepareIndex<br/>解析文档/动态映射]
    ReplicaPath --> PrepareIndex

    PrepareIndex --> Listeners[indexingOperationListeners.preIndex]
    Listeners --> EngineIndex[engine.index<br/>调用InternalEngine]

    EngineIndex --> VersionLock[获取UID锁<br/>versionMap.acquireLock]
    VersionLock --> VersionCheck[版本检查<br/>getUnderLock]

    VersionCheck --> |冲突| VersionConflict[抛出VersionConflictException]
    VersionCheck --> |通过| SeqNoAssign{是Primary?}

    SeqNoAssign --> |Yes| GenerateSeqNo[生成SeqNo<br/>localCheckpoint + 1]
    SeqNoAssign --> |No| UseSeqNo[使用Primary的SeqNo]

    GenerateSeqNo --> Translog[写入Translog<br/>translog.add]
    UseSeqNo --> Translog

    Translog --> Lucene[写入Lucene<br/>IndexWriter.updateDocument]
    Lucene --> UpdateVM[更新LiveVersionMap<br/>putIndexUnderLock]
    UpdateVM --> UpdateLCP[更新LocalCheckpointTracker<br/>markSeqNoAsProcessed]
    UpdateLCP --> ReleaseLock[释放UID锁]
    ReleaseLock --> PostListeners[indexingOperationListeners.postIndex]
    PostListeners --> End[返回IndexResult]

    style EngineIndex fill:#e1f5ff
    style Translog fill:#ffe1e1
    style Lucene fill:#fff4e1
    style UpdateVM fill:#e1ffe1

// IndexShard.java: 953-976行
public Engine.IndexResult applyIndexOperationOnPrimary(
    long version,
    VersionType versionType,
    SourceToParse sourceToParse,
    long ifSeqNo,
    long ifPrimaryTerm,
    long autoGeneratedTimestamp,
    boolean isRetry
) throws IOException {
    // 调用通用索引方法,标记为PRIMARY来源
    return applyIndexOperation(
        getEngine(),
        UNASSIGNED_SEQ_NO,      // Primary自己分配seqNo
        getOperationPrimaryTerm(),
        version,
        versionType,
        ifSeqNo,
        ifPrimaryTerm,
        autoGeneratedTimestamp,
        isRetry,
        Engine.Operation.Origin.PRIMARY,  // 标记为Primary操作
        sourceToParse
    );
}

// IndexShard.java: 1001-1046行
private Engine.IndexResult applyIndexOperation(
    Engine engine,
    long seqNo,
    long opPrimaryTerm,
    long version,
    VersionType versionType,
    // ... 其他参数
) throws IOException {
    // 检查是否允许写入
    ensureWriteAllowed(origin);

    Engine.Index operation;
    try {
        // 准备索引操作:解析文档、处理动态映射
        operation = prepareIndex(
            mapperService,
            sourceToParse,
            seqNo,
            opPrimaryTerm,
            version,
            versionType,
            origin,
            autoGeneratedTimeStamp,
            isRetry,
            ifSeqNo,
            ifPrimaryTerm,
            getRelativeTimeInNanos()
        );

        // 检查是否需要动态映射更新
        Mapping update = operation.parsedDoc().dynamicMappingsUpdate();
        if (update != null) {
            // 返回需要映射更新的结果,由上层处理
            return new Engine.IndexResult(update, operation.parsedDoc().id());
        }
    } catch (Exception e) {
        // 文档级别失败(解析错误、映射失败等)
        verifyNotClosed(e);
        return new Engine.IndexResult(e, version, opPrimaryTerm, seqNo, sourceToParse.id());
    }

    // 执行索引操作
    return index(engine, operation);
}

// IndexShard.java: 1116-1133行
private Engine.IndexResult index(Engine engine, Engine.Index index) throws IOException {
    try {
        final Engine.IndexResult result;
        // 执行前置监听器(统计、日志等)
        final Engine.Index preIndex = indexingOperationListeners.preIndex(shardId, index);

        try {
            if (logger.isTraceEnabled()) {
                logger.trace(
                    "index [{}] seq# [{}] allocation-id [{}] primaryTerm [{}] operationPrimaryTerm [{}] origin [{}]",
                    preIndex.id(),
                    preIndex.seqNo(),
                    routingEntry().allocationId(),
                    preIndex.primaryTerm(),
                    getOperationPrimaryTerm(),
                    preIndex.origin()
                );
            }

            // 调用Engine的index方法
            result = engine.index(preIndex);

            // 日志记录结果
            if (logger.isTraceEnabled()) {
                logger.trace("index-done [{}] seq# [{}] result-seq# [{}] result-term [{}] failure [{}]",
                    preIndex.id(), preIndex.seqNo(), result.getSeqNo(), result.getTerm(), result.getFailure());
            }
        } catch (Exception e) {
            // 异常处理...
        }
    } finally {
        // 执行后置监听器
    }
    return result;
}

// InternalEngine.java: 1175-1309行
@Override
public IndexResult index(Index index) throws IOException {
    final boolean doThrottle = index.origin().isRecovery() == false;
    try (var ignored1 = acquireEnsureOpenRef()) {
        assert assertIncomingSequenceNumber(index.origin(), index.seqNo());
        int reservedDocs = 0;
        try (
            // 获取文档级别的锁(基于UID)
            Releasable ignored = versionMap.acquireLock(index.uid());
            // 如果需要,获取限流令牌
            Releasable indexThrottle = doThrottle ? throttle.acquireThrottle() : () -> {}
        ) {
            lastWriteNanos = index.startTime();

            /*
             * 自动生成ID优化:
             * 如果文档ID是自动生成的且无重试,可以直接使用addDocument
             * 而不是updateDocument,避免版本检查和查找开销
             */
            final IndexingStrategy plan;

            // 第一步:制定索引策略(版本检查、冲突检测)
            if (index.origin() == Operation.Origin.PRIMARY) {
                plan = planIndexingAsPrimary(index);
            } else {
                plan = planIndexingAsNonPrimary(index);
            }

            final IndexResult indexResult;
            // 如果策略指示有预检错误,直接返回
            if (plan.earlyResultOnPreflightError.isPresent()) {
                indexResult = plan.earlyResultOnPreflightError.get();
            } else {
                // 第二步:写入Lucene
                if (plan.indexIntoLucene || plan.addStaleOpToLucene) {
                    indexResult = indexIntoLucene(index, plan);
                } else {
                    // 不需要写Lucene(例如已存在的更老版本)
                    indexResult = new IndexResult(
                        plan.versionForIndexing,
                        index.primaryTerm(),
                        index.seqNo(),
                        plan.currentNotFoundOrDeleted,
                        index.id()
                    );
                }
            }

            // 第三步:写入Translog(如果不是来自Translog)
            if (index.origin().isFromTranslog() == false) {
                final Translog.Location location;
                if (indexResult.getResultType() == Result.Type.SUCCESS) {
                    // 成功:写入Index操作
                    location = translog.add(new Translog.Index(index, indexResult));
                } else if (indexResult.getSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
                    // 失败但有seqNo:写入NoOp
                    final NoOp noOp = new NoOp(
                        indexResult.getSeqNo(),
                        index.primaryTerm(),
                        index.origin(),
                        index.startTime(),
                        indexResult.getFailure().toString()
                    );
                    location = innerNoOp(noOp).getTranslogLocation();
                } else {
                    location = null;
                }
                indexResult.setTranslogLocation(location);
            }

            // 第四步:更新LiveVersionMap
            if (plan.indexIntoLucene && indexResult.getResultType() == Result.Type.SUCCESS) {
                final Translog.Location translogLocation =
                    trackTranslogLocation.get() ? indexResult.getTranslogLocation() : null;
                versionMap.maybePutIndexUnderLock(
                    index.uid(),
                    new IndexVersionValue(
                        translogLocation,       // Translog位置(用于实时GET)
                        plan.versionForIndexing,
                        index.seqNo(),
                        index.primaryTerm()
                    )
                );
            }

            // 第五步:更新LocalCheckpointTracker
            localCheckpointTracker.markSeqNoAsProcessed(indexResult.getSeqNo());
            if (indexResult.getTranslogLocation() == null) {
                // 来自Translog或无seqNo,标记为已持久化
                localCheckpointTracker.markSeqNoAsPersisted(indexResult.getSeqNo());
            }

            indexResult.setTook(relativeTimeInNanosSupplier.getAsLong() - index.startTime());
            indexResult.freeze();
            return indexResult;
        } finally {
            releaseInFlightDocs(reservedDocs);
        }
    } catch (RuntimeException | IOException e) {
        // 引擎失败处理
        try {
            if (e instanceof AlreadyClosedException == false && treatDocumentFailureAsTragicError(index)) {
                failEngine("index id[" + index.id() + "] origin[" + index.origin() + "] seq#[" + index.seqNo() + "]", e);
            } else {
                maybeFailEngine("index id[" + index.id() + "] origin[" + index.origin() + "] seq#[" + index.seqNo() + "]", e);
            }
        } catch (Exception inner) {
            e.addSuppressed(inner);
        }
        throw e;
    }
}

// InternalEngine.java: 1356-1408行
private IndexingStrategy planIndexingAsPrimary(Index index) throws IOException {
    assert index.origin() == Operation.Origin.PRIMARY : "planing as primary but origin isn't. got " + index.origin();

    // 自动生成ID且非重试:可能使用优化路径
    final boolean canOptimizeAddDocument = index.isRetry() == false
        && index.autoGeneratedIdTimestamp() != IndexRequest.UNSET_AUTO_GENERATED_TIMESTAMP;

    if (canOptimizeAddDocument) {
        // 检查是否真的可以优化(没有重复风险)
        if (mayHaveBeenIndexedBefore(index)) {
            // 有重复风险,使用普通路径
            versionMap.enforceSafeAccess();
        } else {
            // 可以优化:直接addDocument,无需版本检查
            return IndexingStrategy.optimizedAppendOnly(index.version(), 0);
        }
    } else {
        // 非自动ID,总是需要版本检查
        versionMap.enforceSafeAccess();
    }

    // 从VersionMap获取当前版本
    final OpVsLuceneDocStatus opVsLucene;
    final VersionValue versionValue = versionMap.getUnderLock(index.uid());

    if (versionValue != null) {
        // 存在版本信息:检查冲突
        final long currentVersion = versionValue.version;
        final VersionType versionType = index.versionType();

        // 版本冲突检测
        if (versionType.isVersionConflictForWrites(
            currentVersion,
            index.version(),
            versionValue.isDelete()
        )) {
            // 抛出版本冲突异常
            final String message = versionType.explainConflictForWrites(
                currentVersion,
                index.version(),
                versionValue.isDelete()
            );
            return IndexingStrategy.failWithVersionConflict(
                new VersionConflictEngineException(shardId, index, currentVersion, versionValue.isDelete(), message),
                currentVersion,
                0
            );
        }

        // 检查if_seq_no和if_primary_term条件
        if (index.getIfSeqNo() != SequenceNumbers.UNASSIGNED_SEQ_NO) {
            // Compare-and-swap检查
            final boolean samePrimaryTerm = index.getIfPrimaryTerm() == versionValue.term;
            final boolean sameSeqNo = index.getIfSeqNo() == versionValue.seqNo;
            if (samePrimaryTerm == false || sameSeqNo == false) {
                return IndexingStrategy.failWithVersionConflict(...);
            }
        }

        // 通过版本检查
        opVsLucene = compareOpToVersionMapOnSeqNo(
            index.id(),
            index.seqNo(),
            index.primaryTerm(),
            versionValue
        );
    } else {
        // 没有版本信息:检查Lucene
        opVsLucene = compareOpToLuceneDocBasedOnVersions(index);
    }

    // 构建索引策略
    return IndexingStrategy.processNormally(
        opVsLucene == OpVsLuceneDocStatus.LUCENE_DOC_NOT_FOUND,
        versionType.updateVersion(currentVersion, index.version()),
        0
    );
}

flowchart TB
    Start[开始: index操作] --> VersionCheck[步骤1: 版本检查<br/>LiveVersionMap]
    VersionCheck --> |冲突| Reject[拒绝操作]
    VersionCheck --> |通过| AssignSeqNo[步骤2: 分配SeqNo<br/>Primary生成]
    AssignSeqNo --> WriteTranslog[步骤3: 写入Translog<br/>WAL保证]
    WriteTranslog --> WriteLucene[步骤4: 写入Lucene<br/>IndexWriter]
    WriteLucene --> UpdateVersionMap[步骤5: 更新VersionMap<br/>记录版本]
    UpdateVersionMap --> UpdateCheckpoint[步骤6: 更新Checkpoint<br/>LocalCheckpointTracker]
    UpdateCheckpoint --> End[完成]

    style WriteTranslog fill:#ffe1e1
    style WriteLucene fill:#fff4e1
    style UpdateVersionMap fill:#e1ffe1

flowchart TB
    Start[触发源<br/>定时/手动/索引缓冲满] --> Scheduler[RefreshScheduler<br/>或手动调用]
    Scheduler --> EngineRefresh[InternalEngine.refresh]

    EngineRefresh --> GetRM[获取ReferenceManager<br/>External/Internal]
    GetRM --> ReadLock[获取Engine读锁<br/>防止Engine关闭]

    ReadLock --> CheckScope{Blocking?}
    CheckScope --> |Yes| BlockingRefresh[maybeRefreshBlocking<br/>同步刷新]
    CheckScope --> |No| MaybeRefresh[maybeRefresh<br/>尝试刷新]

    BlockingRefresh --> IWGetReader[IndexWriter.getReader<br/>或commit]
    MaybeRefresh --> IWGetReader

    IWGetReader --> FlushBuffer[刷新内存缓冲<br/>生成Segment]
    FlushBuffer --> CreateReader[创建新DirectoryReader]
    CreateReader --> UpdateRM[更新ReferenceManager<br/>原子切换Reader]

    UpdateRM --> CloseOld[关闭旧Reader<br/>RefCount减1]
    CloseOld --> TriggerListener[触发RefreshListener]

    TriggerListener --> PruneVM[清理LiveVersionMap<br/>pruneVersionMap]
    PruneVM --> UpdateCheckpoint[更新refreshedCheckpoint]
    UpdateCheckpoint --> ReleaseLock[释放读锁]
    ReleaseLock --> End[完成]

    style IWGetReader fill:#fff4e1
    style CreateReader fill:#fff4e1
    style PruneVM fill:#e1ffe1

// InternalEngine.java: 2074-2137行
protected final RefreshResult refresh(String source, SearcherScope scope, boolean block) throws EngineException {
    // 记录refresh前的本地检查点
    final long localCheckpointBeforeRefresh = localCheckpointTracker.getProcessedCheckpoint();
    boolean refreshed = false;
    long segmentGeneration = RefreshResult.UNKNOWN_GENERATION;

    try {
        // 增加store引用计数,防止关闭
        if (store.tryIncRef()) {
            try {
                // 获取对应的ReferenceManager (External或Internal)
                ReferenceManager<ElasticsearchDirectoryReader> referenceManager =
                    getReferenceManager(scope);
                long generationBeforeRefresh = lastCommittedSegmentInfos.getGeneration();

                // 获取Engine读锁,防止与Engine关闭并发
                final var engineReadLock = engineConfig.getEngineResetLock().readLock();

                if (block) {
                    // 同步刷新:阻塞直到完成
                    engineReadLock.lock();
                    try {
                        referenceManager.maybeRefreshBlocking();
                        segmentGeneration = segmentGenerationAfterRefresh(
                            referenceManager,
                            generationBeforeRefresh
                        );
                        refreshed = true;
                    } finally {
                        engineReadLock.unlock();
                    }
                } else {
                    // 异步刷新:尝试获取锁
                    if (engineReadLock.tryLock()) {
                        try {
                            refreshed = referenceManager.maybeRefresh();
                            if (refreshed) {
                                segmentGeneration = segmentGenerationAfterRefresh(
                                    referenceManager,
                                    generationBeforeRefresh
                                );
                            }
                        } finally {
                            engineReadLock.unlock();
                        }
                    }
                }
            } finally {
                store.decRef();
            }

            // 更新已刷新的检查点
            if (refreshed) {
                lastRefreshedCheckpointListener.updateRefreshedCheckpoint(
                    localCheckpointBeforeRefresh
                );
            }
        } else {
            refreshed = false;
        }
    } catch (AlreadyClosedException e) {
        failOnTragicEvent(e);
        throw e;
    } catch (Exception e) {
        try {
            failEngine("refresh failed source[" + source + "]", e);
        } catch (Exception inner) {
            e.addSuppressed(inner);
        }
        throw new RefreshFailedEngineException(shardId, e);
    }

    return new RefreshResult(refreshed, segmentGeneration);
}

// Lucene ReferenceManager的核心方法
public final boolean maybeRefresh() throws IOException {
    ensureOpen();

    // 获取最新的Reader
    final G refreshedReference = refreshIfNeeded(current);

    if (refreshedReference != null) {
        assert refreshedReference != current;

        // 原子切换到新Reader
        boolean success = false;
        try {
            // 通知监听器:beforeRefresh
            for (RefreshListener listener : refreshListeners) {
                listener.beforeRefresh();
            }

            // 原子替换
            release(current);
            current = refreshedReference;
            success = true;

            // 通知监听器:afterRefresh
            for (RefreshListener listener : refreshListeners) {
                listener.afterRefresh(true);
            }
        } finally {
            if (!success) {
                release(refreshedReference);
            }
        }
        return true;
    }

    return false;
}

// InternalEngine内的RefreshListener实现
private final class LastRefreshedCheckpointListener implements ReferenceManager.RefreshListener {
    final AtomicLong refreshedCheckpoint;
    private long pendingCheckpoint;

    @Override
    public void beforeRefresh() {
        // Refresh前:记录当前的处理检查点
        // 所有 <= pendingCheckpoint 的操作都将在Refresh后可见
        pendingCheckpoint = localCheckpointTracker.getProcessedCheckpoint();
    }

    @Override
    public void afterRefresh(boolean didRefresh) {
        if (didRefresh) {
            // Refresh后:更新已刷新的检查点
            updateRefreshedCheckpoint(pendingCheckpoint);

            // 清理LiveVersionMap中已刷到Lucene的条目
            pruneVersionMap(pendingCheckpoint);
        }
    }

    void pruneVersionMap(long refreshedSeqNo) {
        // 遍历VersionMap,清理已持久化到Lucene的Location
        versionMap.pruneTombstones(refreshedSeqNo, relativeTimeInNanosSupplier.getAsLong());

        // 对于 seqNo <= refreshedSeqNo 的IndexVersionValue:
        // - 清除translogLocation字段(因为已经在Lucene中)
        // - 保留version/seqNo/term(仍需用于版本检查)
    }
}

flowchart TB
    Start[触发源<br/>Translog满/定时] --> FlushPolicy[FlushPolicy<br/>shouldPeriodicallyFlush]
    FlushPolicy --> EngineFlush[InternalEngine.flush]

    EngineFlush --> AcquireLock[获取flushLock<br/>防止并发Flush]
    AcquireLock --> CheckCondition{需要Flush?}

    CheckCondition --> |No| SkipFlush[跳过Flush]
    CheckCondition --> |Yes| PreRefresh[步骤1: Refresh<br/>刷新到新段]

    PreRefresh --> RollTranslog[步骤2: 滚动Translog<br/>translog.rollGeneration]
    RollTranslog --> PrepareCommit[步骤3: 准备提交<br/>IndexWriter.prepareCommit]

    PrepareCommit --> WriteSegmentsN[写入segments_N文件<br/>包含提交元数据]
    WriteSegmentsN --> FinishCommit[步骤4: 完成提交<br/>IndexWriter.finishCommit]

    FinishCommit --> Fsync[步骤5: Fsync所有文件<br/>directory.sync]
    Fsync --> TrimTranslog[步骤6: 清理Translog<br/>trimUnreferencedReaders]

    TrimTranslog --> DeleteOld[删除旧generation<br/>translog-1.tlog...]
    DeleteOld --> UpdateMetrics[更新统计信息<br/>lastFlushTimestamp]
    UpdateMetrics --> ReleaseLock[释放flushLock]
    ReleaseLock --> Notify[通知FlushListener]
    Notify --> End[完成]

    style RollTranslog fill:#ffe1e1
    style WriteSegmentsN fill:#fff4e1
    style Fsync fill:#e1f5ff

// InternalEngine.java: 2213-2242行
private boolean shouldPeriodicallyFlush(long flushThresholdSizeInBytes, long flushThresholdAgeInNanos) {
    ensureOpen();

    // 检查是否在大合并后需要Flush
    if (shouldPeriodicallyFlushAfterBigMerge.get()) {
        return true;
    }

    // 获取上次提交的本地检查点
    final long localCheckpointOfLastCommit = Long.parseLong(
        lastCommittedSegmentInfos.userData.get(SequenceNumbers.LOCAL_CHECKPOINT_KEY)
    );

    // 计算需要的Translog generation
    final long translogGenerationOfLastCommit =
        translog.getMinGenerationForSeqNo(localCheckpointOfLastCommit + 1)
            .translogFileGeneration();

    // 检查Translog大小和时间
    if (translog.sizeInBytesByMinGen(translogGenerationOfLastCommit) < flushThresholdSizeInBytes
        && relativeTimeInNanosSupplier.getAsLong() - lastFlushTimestamp < flushThresholdAgeInNanos) {
        // 既未超过大小阈值,也未超过时间阈值
        return false;
    }

    /*
     * 防止无限Flush循环:
     * 只有当新提交会指向更新的generation时,才触发Flush
     * 这确保Flush后Translog大小会实际减小
     */
    final long translogGenerationOfNewCommit = translog.getMinGenerationForSeqNo(
        localCheckpointTracker.getProcessedCheckpoint() + 1
    ).translogFileGeneration();

    return translogGenerationOfNewCommit > translogGenerationOfLastCommit
        || localCheckpointTracker.getProcessedCheckpoint() == localCheckpointTracker.getMaxSeqNo();
}

// InternalEngine.java: 2283-2330行
protected void flushHoldingLock(boolean force, boolean waitIfOngoing, ActionListener<FlushResult> listener) {
    // 获取Flush锁
    try {
        flushLock.lock();

        // 检查是否有未提交的更改
        boolean hasUncommittedChanges = hasUncommittedChanges();

        if (hasUncommittedChanges
            || force
            || shouldPeriodicallyFlush()
            || getProcessedLocalCheckpoint() > Long.parseLong(
                lastCommittedSegmentInfos.userData.get(SequenceNumbers.LOCAL_CHECKPOINT_KEY)
            )) {

            ensureCanFlush();
            Translog.Location commitLocation = getTranslogLastWriteLocation();

            try {
                // 步骤1: 滚动Translog generation
                translog.rollGeneration();
                logger.trace("starting commit for flush; commitTranslog=true");

                long lastFlushTimestamp = relativeTimeInNanosSupplier.getAsLong();

                // 预先记录即将生成的segment generation
                preCommitSegmentGeneration.set(lastCommittedSegmentInfos.getGeneration() + 1);

                // 步骤2: 提交IndexWriter
                commitIndexWriter(indexWriter, translog);
                logger.trace("finished commit for flush");

                // 步骤3: Refresh以清理VersionMap
                refresh("version_table_flush", SearcherScope.INTERNAL, true);

                // 步骤4: 清理旧的Translog readers
                translog.trimUnreferencedReaders();

                // 更新Translog位置(如果有新写入)
                Translog.Location newLocation = getTranslogLastWriteLocation();
                if (newLocation.compareTo(commitLocation) > 0) {
                    commitLocation = newLocation;
                }
            } catch (Exception e) {
                throw new FlushFailedEngineException(shardId, e);
            } finally {
                this.lastFlushTimestamp = lastFlushTimestamp;
            }

            // 异步确保Translog同步
            asyncEnsureTranslogSynced(commitLocation, new ActionListener<Void>() {
                @Override
                public void onResponse(Void aVoid) {
                    listener.onResponse(new FlushResult(true, generation));
                }

                @Override
                public void onFailure(Exception e) {
                    listener.onFailure(e);
                }
            });
        } else {
            // 无需Flush
            listener.onResponse(new FlushResult(false, generation));
        }
    } finally {
        flushLock.unlock();
    }
}

// InternalEngine.java
private void commitIndexWriter(IndexWriter writer, Translog translog) throws IOException {
    ensureCanFlush();

    try {
        // 获取最新的检查点信息
        final long localCheckpoint = localCheckpointTracker.getProcessedCheckpoint();
        final long maxSeqNo = localCheckpointTracker.getMaxSeqNo();

        // 构建提交的元数据
        final Map<String, String> commitData = new HashMap<>(8);
        commitData.put(SequenceNumbers.LOCAL_CHECKPOINT_KEY, Long.toString(localCheckpoint));
        commitData.put(SequenceNumbers.MAX_SEQ_NO, Long.toString(maxSeqNo));
        commitData.put(Translog.TRANSLOG_UUID_KEY, translog.getTranslogUUID());
        commitData.put(Translog.TRANSLOG_GENERATION_KEY, Long.toString(translog.currentFileGeneration()));
        commitData.put(Engine.HISTORY_UUID_KEY, historyUUID);
        commitData.put(Engine.MIN_RETAINED_SEQNO, Long.toString(softDeletesPolicy.getMinRetainedSeqNo()));

        if (forceMergeUUID != null) {
            commitData.put(FORCE_MERGE_UUID_KEY, forceMergeUUID);
        }

        // 设置提交数据
        writer.setLiveCommitData(commitData.entrySet());

        // 步骤1: prepareCommit (写入segments_N但不fsync)
        writer.prepareCommit();

        // 步骤2: finishCommit (重命名segments_N使其可见)
        writer.commit();

    } catch (final Exception ex) {
        try {
            failEngine("lucene commit failed", ex);
        } catch (final Exception inner) {
            ex.addSuppressed(inner);
        }
        throw ex;
    } catch (final AssertionError e) {
        // IndexWriter bug导致的断言失败视为致命错误
        throw new FlushFailedEngineException(shardId, e);
    }
}

// Translog.java: 滚动generation
public void rollGeneration() throws IOException {
    writeLock.lock();
    try {
        ensureOpen();

        // 关闭当前writer
        try {
            final TranslogWriter newWriter = createWriter(
                current.getGeneration() + 1,
                getMinFileGeneration(),
                current.getLastSyncedCheckpoint().offset
            );

            // 将当前writer移到readers列表
            readers.add(current.closeIntoReader());
            current = newWriter;

        } catch (final Exception e) {
            closeOnTragicEvent(e);
            throw new TranslogException(shardId, "failed to roll translog", e);
        }
    } finally {
        writeLock.unlock();
    }
}

// 清理不再需要的readers
public void trimUnreferencedReaders() throws IOException {
    writeLock.lock();
    try {
        // 获取最小需要保留的generation
        long minReferencedGen = deletionPolicy.getMinTranslogGenerationForRecovery();

        // 删除更老的generation文件
        for (Iterator<TranslogReader> it = readers.iterator(); it.hasNext(); ) {
            TranslogReader reader = it.next();
            if (reader.getGeneration() < minReferencedGen) {
                it.remove();
                IOUtils.closeWhileHandlingException(reader);
                Files.deleteIfExists(reader.path());
                logger.trace("deleted translog [{}]", reader.path());
            }
        }
    } finally {
        writeLock.unlock();
    }
}

flowchart TB
    Start[DELETE /index/_doc/1] --> IndexShard[IndexShard.applyDeleteOperationOnPrimary]
    IndexShard --> EngineDelete[InternalEngine.delete]

    EngineDelete --> AcquireLock[获取UID锁<br/>versionMap.acquireLock]
    AcquireLock --> VersionCheck[版本检查<br/>getUnderLock]

    VersionCheck --> |冲突| Conflict[版本冲突异常]
    VersionCheck --> |通过| AssignSeqNo[分配SeqNo]

    AssignSeqNo --> CreateTombstone[创建Tombstone文档<br/>ParsedDocument.deleteTombstone]
    CreateTombstone --> SetFields[设置字段<br/>seqNo/primaryTerm/version]

    SetFields --> AddSoftDelete[添加软删除字段<br/>_soft_deletes: true]
    AddSoftDelete --> SoftUpdate[Lucene软删除<br/>softUpdateDocument]

    SoftUpdate --> WriteTranslog[写入Translog<br/>Translog.Delete]
    WriteTranslog --> UpdateVM[更新VersionMap<br/>putDeleteUnderLock]

    UpdateVM --> UpdateLCP[更新LocalCheckpointTracker]
    UpdateLCP --> ReleaseLock[释放UID锁]
    ReleaseLock --> End[返回DeleteResult]

    style CreateTombstone fill:#fff4e1
    style SoftUpdate fill:#fff4e1
    style UpdateVM fill:#e1ffe1

// InternalEngine.java: 1633-1717行
@Override
public DeleteResult delete(Delete delete) throws IOException {
    // 强制使用安全访问(需要版本检查)
    versionMap.enforceSafeAccess();
    assert assertIncomingSequenceNumber(delete.origin(), delete.seqNo());

    final DeleteResult deleteResult;
    int reservedDocs = 0;

    try (
        var ignored = acquireEnsureOpenRef();
        Releasable ignored2 = versionMap.acquireLock(delete.uid())
    ) {
        lastWriteNanos = delete.startTime();

        // 第一步:制定删除策略
        final DeletionStrategy plan = deletionStrategyForOperation(delete);
        reservedDocs = plan.reservedDocs;

        if (plan.earlyResultOnPreflightError.isPresent()) {
            // 预检失败(例如版本冲突)
            assert delete.origin() == Operation.Origin.PRIMARY : delete.origin();
            deleteResult = plan.earlyResultOnPreflightError.get();
        } else {
            // 第二步:写入Lucene (Soft Delete)
            if (plan.deleteFromLucene) {
                deleteResult = deleteInLucene(delete, plan);
            } else {
                // 不需要写Lucene(已经删除或旧版本)
                deleteResult = new DeleteResult(
                    plan.versionOfDeletion,
                    delete.primaryTerm(),
                    delete.seqNo(),
                    plan.currentlyDeleted == false,
                    delete.id()
                );
            }
        }

        // 第三步:写入Translog
        if (delete.origin().isFromTranslog() == false && deleteResult.getResultType() == Result.Type.SUCCESS) {
            final Translog.Location location = translog.add(new Translog.Delete(delete, deleteResult));
            deleteResult.setTranslogLocation(location);
        }

        // 第四步:更新LocalCheckpointTracker
        localCheckpointTracker.markSeqNoAsProcessed(deleteResult.getSeqNo());
        if (deleteResult.getTranslogLocation() == null) {
            localCheckpointTracker.markSeqNoAsPersisted(deleteResult.getSeqNo());
        }

        deleteResult.setTook(System.nanoTime() - delete.startTime());
        deleteResult.freeze();
    } catch (RuntimeException | IOException e) {
        try {
            maybeFailEngine("delete", e);
        } catch (Exception inner) {
            e.addSuppressed(inner);
        }
        throw e;
    } finally {
        releaseInFlightDocs(reservedDocs);
    }

    // 定期清理删除的Tombstone
    maybePruneDeletes();
    return deleteResult;
}

// InternalEngine.java: 1840-1876行
private DeleteResult deleteInLucene(Delete delete, DeletionStrategy plan) throws IOException {
    assert assertMaxSeqNoOfUpdatesIsAdvanced(delete.uid(), delete.seqNo(), false, false);

    try {
        // 创建删除Tombstone文档
        final ParsedDocument tombstone = ParsedDocument.deleteTombstone(
            engineConfig.getIndexSettings().seqNoIndexOptions(),
            delete.id()
        );
        assert tombstone.docs().size() == 1 : "Tombstone doc should have single doc [" + tombstone + "]";

        // 设置seqNo和primaryTerm
        tombstone.updateSeqID(delete.seqNo(), delete.primaryTerm());

        // 设置version
        tombstone.version().setLongValue(plan.versionOfDeletion);

        final LuceneDocument doc = tombstone.docs().get(0);

        // 确保有_tombstone字段
        assert doc.getField(SeqNoFieldMapper.TOMBSTONE_NAME) != null
            : "Delete tombstone document but _tombstone field is not set [" + doc + " ]";

        // 添加软删除字段
        doc.add(softDeletesField);

        // 写入Lucene
        if (plan.addStaleOpToLucene || plan.currentlyDeleted) {
            // 文档不存在或已删除:直接addDocument
            indexWriter.addDocument(doc);
        } else {
            // 文档存在:使用softUpdateDocument
            // 先添加新的tombstone,再标记旧文档为软删除
            indexWriter.softUpdateDocument(
                new Term(IdFieldMapper.NAME, delete.uid()),
                doc,
                softDeletesField
            );
        }

        return new DeleteResult(
            plan.versionOfDeletion,
            delete.primaryTerm(),
            delete.seqNo(),
            plan.currentlyDeleted == false,  // found
            delete.id()
        );
    } catch (final Exception ex) {
        // 删除失败视为致命错误
        if (ex instanceof AlreadyClosedException == false && ex instanceof IOException == false) {
            throw new AssertionError("Unexpected exception during deletion", ex);
        }
        throw ex;
    }
}