// RestCreateSnapshotAction.java
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    // 解析请求参数 - 提取仓库名和快照名等信息
    String repository = request.param("repository");
    String snapshot = request.param("snapshot");
    
    // 构建快照请求对象 - 创建 CreateSnapshotRequest
    CreateSnapshotRequest createSnapshotRequest = new CreateSnapshotRequest(repository, snapshot);
    request.applyContentParser(p -> createSnapshotRequest.source(p.mapOrdered()));
    
    // 设置超时时间和等待完成标志
    createSnapshotRequest.masterNodeTimeout(request.paramAsTime("master_timeout", createSnapshotRequest.masterNodeTimeout()));
    createSnapshotRequest.waitForCompletion(request.paramAsBoolean("wait_for_completion", false));
    
    // 客户端调用 - 转发到集群管理接口
    return channel -> client.admin().cluster().createSnapshot(createSnapshotRequest, new RestToXContentListener<>(channel));
}

// TransportCreateSnapshotAction.java
@Override
protected void masterOperation(
    Task task,
    final CreateSnapshotRequest request,
    ClusterState state,
    final ActionListener<CreateSnapshotResponse> listener
) {
    // 检查等待完成标志 - 决定同步还是异步执行
    if (request.waitForCompletion()) {
        // 同步执行 - 执行并等待完成，阻塞直到快照完成
        snapshotsService.executeSnapshot(request, listener.map(CreateSnapshotResponse::new));
    } else {
        // 异步执行 - 创建快照任务，立即返回响应
        snapshotsService.createSnapshot(request, listener.map(snapshot -> new CreateSnapshotResponse()));
    }
}

// SnapshotsService.java
public void createSnapshot(final CreateSnapshotRequest request, final ActionListener<Snapshot> listener) {
    final String repositoryName = request.repository();
    final String snapshotName = IndexNameExpressionResolver.resolveDateMathExpression(request.snapshot());
    
    // 验证仓库名和快照名
    validate(repositoryName, snapshotName);
    
    // 生成新的快照UUID
    final SnapshotId snapshotId = new SnapshotId(snapshotName, UUIDs.randomBase64UUID());
    Repository repository = repositoriesService.repository(request.repository());
    
    // 检查仓库是否只读
    if (repository.isReadOnly()) {
        listener.onFailure(new RepositoryException(repository.getMetadata().name(), "cannot create snapshot in a readonly repository"));
        return;
    }
    
    // 提交快照创建请求 - 获取仓库数据并准备任务
    submitCreateSnapshotRequest(request, listener, repository, new Snapshot(repositoryName, snapshotId), repository.getMetadata());
}

// SnapshotsService.SnapshotTaskExecutor.createSnapshot
private SnapshotsInProgress createSnapshot(
    CreateSnapshotTask createSnapshotTask,
    TaskContext<SnapshotTask> taskContext,
    ClusterState currentState,
    SnapshotsInProgress snapshotsInProgress
) {
    final RepositoryData repositoryData = createSnapshotTask.repositoryData;
    final Snapshot snapshot = createSnapshotTask.snapshot;
    final String repositoryName = snapshot.getRepository();
    final String snapshotName = snapshot.getSnapshotId().getName();
    
    // 1. 基础验证 - 验证请求并生成快照元数据
    ensureRepositoryExists(repositoryName, currentState);
    ensureSnapshotNameAvailableInRepo(repositoryData, snapshotName, repository);
    ensureSnapshotNameNotRunning(snapshotsInProgress, repositoryName, snapshotName);
    validate(repositoryName, snapshotName, currentState);
    
    // 2. 并发限制检查
    final SnapshotDeletionsInProgress deletionsInProgress = SnapshotDeletionsInProgress.get(currentState);
    ensureNoCleanupInProgress(currentState, repositoryName, snapshotName, "create snapshot");
    ensureBelowConcurrencyLimit(repositoryName, snapshotName, snapshotsInProgress, deletionsInProgress);
    
    // 3. 解析要快照的索引
    Map<Boolean, List<String>> requestedIndices = Arrays.stream(
        indexNameExpressionResolver.concreteIndexNames(currentState, request)
    ).collect(Collectors.partitioningBy(systemIndices::isSystemIndex));
    
    // 4. 创建分片映射
    final Map<String, IndexId> indexIds = repositoryData.resolveNewIndices(indices, allIndices);
    ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards = shards(
        snapshotsInProgress,
        deletionsInProgress,
        currentState,
        indexIds.values(),
        useShardGenerations(version),
        repositoryData,
        repositoryName
    );
    
    // 5. 生成快照进度条目 - 创建 SnapshotsInProgress.Entry
    final var newEntry = SnapshotsInProgress.startedEntry(
        snapshot,
        request.includeGlobalState(),
        request.partial(),
        indexIds,
        dataStreams,
        featureStates,
        repositoryData.getGenId(),
        shards,
        version,
        startTime,
        repositoryData.getUuid(),
        request.userMetadata()
    );
    
    // 6. 更新集群状态 - 将快照信息加入集群状态
    final var res = snapshotsInProgress.withAddedEntry(newEntry);
    taskContext.success(() -> {
        logger.info("snapshot [{}] started", snapshot);
        createSnapshotTask.listener.onResponse(snapshot);
        if (newEntry.state().completed()) {
            endSnapshot(newEntry, currentState.metadata(), createSnapshotTask.repositoryData);
        }
    });
    return res;
}

// MasterService.java
protected void publish(
    ClusterStatePublicationEvent clusterStatePublicationEvent,
    ClusterStatePublisher.AckListener ackListener,
    ActionListener<Void> publicationListener
) {
    // 广播集群状态 - 通知所有节点快照已开始
    clusterStatePublisher.publish(
        clusterStatePublicationEvent,
        new ThreadedActionListener<>(
            threadPoolExecutor,
            new ContextPreservingActionListener<>(ackListener, threadContext)
        ),
        publicationListener
    );
}

// SnapshotShardsService.java
@Override
public void clusterChanged(ClusterChangedEvent event) {
    try {
        final var currentSnapshots = SnapshotsInProgress.get(event.state());
        // 分片快照服务监听 - 检测到新的快照任务
        if (SnapshotsInProgress.get(event.previousState()).equals(currentSnapshots) == false) {
            final var localNodeId = event.state().nodes().getLocalNodeId();
            final var removingLocalNode = event.nodesDelta().removedNodes().stream().anyMatch(n -> n.getId().equals(localNodeId));
            synchronized (shardSnapshots) {
                cancelRemoved(currentSnapshots);
                for (SnapshotsInProgress.Entry entry : currentSnapshots.forRepo(repositoryName)) {
                    // 启动分片快照任务 - 为每个分片创建快照任务
                    handleUpdatedSnapshotsInProgressEntry(localNodeId, removingLocalNode, entry);
                }
            }
        }
    } catch (Exception e) {
        logger.warn("Failed to update snapshot state ", e);
    }
}

// SnapshotShardsService.java
private void snapshot(
    final ShardId shardId,
    final Snapshot snapshot,
    final IndexId indexId,
    final IndexShardSnapshotStatus snapshotStatus,
    IndexVersion version,
    final long entryStartTime,
    ActionListener<ShardSnapshotResult> resultListener
) {
    ActionListener.run(resultListener, listener -> {
        snapshotStatus.ensureNotAborted();
        final IndexShard indexShard = indicesService.indexServiceSafe(shardId.getIndex()).getShard(shardId.id());
        
        // 1. 验证分片状态 - 确保分片可以进行快照
        if (indexShard.routingEntry().primary() == false) {
            throw new IndexShardSnapshotFailedException(shardId, "snapshot should be performed only on primary");
        }
        
        // 2. 获取仓库和索引提交点
        final Repository repository = repositoriesService.repository(snapshot.getRepository());
        SnapshotIndexCommit snapshotIndexCommit = null;
        try {
            snapshotIndexCommit = new SnapshotIndexCommit(indexShard.acquireIndexCommitForSnapshot());
            final var shardStateId = getShardStateId(indexShard, snapshotIndexCommit.indexCommit());
            
            // 3. 调用仓库快照接口 - 委托给具体的存储仓库
            repository.snapshotShard(
                new SnapshotShardContext(
                    indexShard.store(),
                    indexShard.mapperService(),
                    snapshot.getSnapshotId(),
                    indexId,
                    snapshotIndexCommit,
                    shardStateId,
                    snapshotStatus,
                    version,
                    entryStartTime,
                    listener
                )
            );
        } catch (Exception e) {
            // 错误处理...
        }
    });
}

// BlobStoreRepository.java
@Override
public void snapshotShard(SnapshotShardContext context) {
    // 仓库分片快照 - 将任务加入快照队列
    shardSnapshotTaskRunner.enqueueShardSnapshot(context);
}

// 🔒 锁定入口1：主节点任务提交
// SnapshotsService.java
public void createSnapshot(final CreateSnapshotRequest request, final ActionListener<Snapshot> listener) {
    // 提交到主节点任务队列，触发集群状态锁定
    masterServiceTaskQueue.submitTask(
        "create_snapshot [" + snapshot.getSnapshotId().getName() + ']',
        new CreateSnapshotTask(repository, repositoryData, listener, snapshot, request, initialRepositoryMetadata),
        request.masterNodeTimeout()
    );
}

// 🔒 锁定入口2：主节点任务执行器
// MasterService.java
private void runTasks(TaskBatch taskBatch) {
    final ClusterState previousClusterState = state();
    
    // 🔒 关键锁定点：获取主服务互斥锁
    synchronized (mutex) {
        ClusterState newClusterState = previousClusterState;
        
        // 🔒 执行集群状态更新任务（需要写锁）
        for (UpdateTask updateTask : taskBatch.toExecute) {
            // 这里会调用 SnapshotsService.SnapshotTaskExecutor.execute()
            newClusterState = updateTask.execute(newClusterState);
        }
        
        // 🔒 发布新的集群状态
        publishClusterState(taskBatch, newClusterState, previousClusterState);
    }
}

// 🔒 锁定入口1：分片快照触发
// SnapshotShardsService.java
private void snapshot(...) {
    final IndexShard indexShard = indicesService.indexServiceSafe(shardId.getIndex()).getShard(shardId.id());
    
    // 🔒 关键锁定点：获取分片的IndexCommit引用
    snapshotIndexCommit = new SnapshotIndexCommit(indexShard.acquireIndexCommitForSnapshot());
    
    // 委托给仓库层处理
    repository.snapshotShard(new SnapshotShardContext(...));
}

// 🔒 锁定入口2：引擎层IndexCommit锁定
// InternalEngine.java
@Override
public IndexCommitRef acquireLastIndexCommit(final boolean flushFirst) throws EngineException {
    // 🔒 可选的Flush锁定 - 短暂暂停写操作
    if (flushFirst) {
        logger.trace("start flush for snapshot");
        PlainActionFuture<FlushResult> future = new PlainActionFuture<>();
        flush(false, true, future);  // 🔒 这里会暂停写入操作
        future.actionGet();  // 等待flush完成
        logger.trace("finish flush for snapshot");
    }
    
    // 🔒 获取IndexCommit引用，增加引用计数
    return acquireIndexCommitRef(() -> combinedDeletionPolicy.acquireIndexCommit(false));
}

// 🔒 锁定入口3：引用计数保护机制
// CombinedDeletionPolicy.java
public synchronized IndexCommit acquireIndexCommit(boolean acquiringSafeCommit) {
    final IndexCommit snapshotting = acquiringSafeCommit ? safeCommit : lastCommit;
    // 🔒 增加引用计数，防止IndexCommit被删除
    snapshotting.setUserData(snapshotting.getUserData());
    return snapshotting;
}

// 🔒 锁定入口1：存储层元数据访问
// Store.java
public MetadataSnapshot getMetadata(IndexCommit commit, boolean lockDirectory) throws IOException {
    ensureOpen();
    failIfCorrupted();
    
    // 🔒 根据需要选择读锁或写锁
    java.util.concurrent.locks.Lock lock = lockDirectory ? metadataLock.writeLock() : metadataLock.readLock();
    lock.lock();
    try (Closeable ignored = lockDirectory ? directory.obtainLock(IndexWriter.WRITE_LOCK_NAME) : () -> {}) {
        // 🔒 在锁保护下读取元数据
        return MetadataSnapshot.loadFromIndexCommit(commit, directory, logger);
    } catch (CorruptIndexException | IndexFormatTooOldException | IndexFormatTooNewException ex) {
        markStoreCorrupted(ex);
        throw ex;
    } finally {
        lock.unlock();  // 🔒 确保锁被释放
    }
}

// 锁定事件
Time: T0 (快照请求提交)
Lock: ClusterState Write Lock (集群状态写锁)
Scope: 整个集群
Duration: 100-500ms
Impact Level: HIGH（集群级影响）

// 具体影响分析
Blocked Operations:
├── ❌ IndicesService.createIndex()                     // 创建索引被阻塞
├── ❌ IndicesService.deleteIndex()                     // 删除索引被阻塞  
├── ❌ MetadataUpdateSettingsService.updateSettings()   // 更新设置被阻塞
├── ❌ MetadataMappingService.putMapping()             // 更新映射被阻塞
├── ❌ SnapshotsService.createSnapshot()               // 其他快照被阻塞
├── ❌ RestoreService.restoreSnapshot()                // 快照恢复被阻塞
└── ✅ Regular read/write operations                   // 正常读写（无影响）

// 锁定事件
Time: T1 (分片开始快照)
Lock: IndexCommit Reference Lock + Optional Flush Lock  
Scope: 单个分片
Duration: 50-200ms（如果需要flush），否则几毫秒
Impact Level: MEDIUM（分片级写操作可能短暂暂停）

// Flush锁定分析（如果flushFirst=true）
if (flushFirst == true) {
    Duration: 50-200ms（取决于未提交数据量）
    Code: InternalEngine.flushHoldingLock()
    
    Temporarily Blocked Operations:
    ├── ❌ IndexShard.index(IndexRequest)               // 新文档写入暂停
    ├── ❌ IndexShard.delete(DeleteRequest)             // 文档删除暂停
    ├── ❌ IndexShard.update(UpdateRequest)             // 文档更新暂停
    ├── ❌ InternalEngine.forceMerge()                  // 强制合并暂停
    └── ✅ IndexShard.get(GetRequest)                   // 文档读取（无影响）
}

// IndexCommit引用锁定分析（默认情况）
Duration: 整个快照过程（几分钟到几小时）
Impact Level: LOW（主要是保护性锁定，不阻止新操作）
Code: CombinedDeletionPolicy.acquireIndexCommit()

Protected Resources:
├── 🔒 IndexCommit.getFileNames()                      // 文件列表保护
├── 🔒 Directory.openInput(fileName)                   // 文件读取保护  
├── 🔒 DeletionPolicy.onCommit()                       // 防止文件被删除策略清理
└── 🔒 Segment files (.si, .cfs, .cfe等)               // 段文件保护

// ❌ 常见误解
"分片快照期间不能对索引数据进行写入/修改"

// ✅ 实际情况  
"分片快照期间写入/修改操作可以正常进行，只是：
 1. 可选的短暂flush锁定（默认不开启）
 2. IndexCommit引用保护（不阻止新写入）
 3. I/O资源竞争（性能轻微下降）"

// 具体验证代码
public void verifyWriteOperationsDuringSnapshot() {
    // 快照进行中
    SnapshotIndexCommit snapshotCommit = new SnapshotIndexCommit(
        indexShard.acquireIndexCommitForSnapshot()
    );
    
    // ✅ 新的写入操作依然可以执行
    IndexRequest newDoc = new IndexRequest("test").id("new_doc").source("field", "value");
    IndexResult result = indexShard.index(newDoc);  // 成功执行
    
    // ✅ 新的删除操作依然可以执行  
    DeleteRequest deleteRequest = new DeleteRequest("test", "doc_to_delete");
    DeleteResult deleteResult = indexShard.delete(deleteRequest);  // 成功执行
    
    // ✅ 新的数据可以被搜索到
    SearchRequest searchRequest = new SearchRequest("test");
    // 搜索能找到刚刚添加的 new_doc
    
    // 🔒 只是旧的IndexCommit文件被保护不被删除
    // 新的写入会创建新的segments和新的IndexCommit
    // 快照读取的是旧的IndexCommit，新数据在新的IndexCommit中
}