# Distributed Area Internals

The Distributed Area contains indexing and coordination systems.

The index path stretches from the user REST command through shard routing down to each individual shard's translog and storage
engine. Reindexing is effectively reading from a source index and writing to a destination index (perhaps on different nodes).
The coordination side includes cluster coordination, shard allocation, cluster autoscaling stats, task management, and cross
cluster replication. Less obvious coordination systems include networking, the discovery plugin system, the snapshot/restore
logic, and shard recovery.

A guide to the general Elasticsearch components can be found [here](https://github.com/elastic/elasticsearch/blob/main/docs/internal/GeneralArchitectureGuide.md).

# Networking

Every elasticsearch node maintains various networking clients and servers,
protocols, and synchronous/asynchronous handling. Our public docs cover user
facing settings and some internal aspects - [Network Settings](https://www.elastic.co/docs/reference/elasticsearch/configuration-reference/networking-settings).

## HTTP Server

The HTTP Server is a single entry point for all external clients (excluding
cross-cluster communication). Management, ingestion, search, and all other
external operations pass through the HTTP server.

Elasticsearch works over HTTP 1.1 and supports features such as TLS, chunked
transfer encoding, content compression, and pipelining. While attempting to
be HTTP spec compliant, Elasticsearch is not a webserver. ES Supports `GET`
requests with a payload (though some old proxies may drop content) and
`POST` for clients unable to send `GET-with-body`. Requests cannot be cached
by middle boxes.

There is no connection limit, but a limit on payload size exists. The default
maximum payload is 100MB after compression. It's a very large number and almost
never a good target that the client should approach. See
`HttpTransportSettings` class.

Security features, including basic security: authentication(authc),
authorization(authz), Transport Layer Security (TLS) are available in the free
tier and achieved with separate x-pack modules.

The HTTP server provides two options for content processing: full aggregation
and incremental processing. Aggregated content is a preferable choice for small
messages that do not fit for incremental parsing (e.g., JSON). Aggregation has
drawbacks: it requires more memory, which is reserved until all bytes are
received. Concurrent incomplete requests can lead to unbounded memory growth and
potential OOMs. Large delimited content, such as bulk indexing, which is
processed in byte chunks, provides better control over memory usage but is more
complicated for application code.

Incremental bulk indexing includes a back-pressure feature. See `org.
elasticsearch.index.IndexingPressure`. When memory pressure grows high
(`LOW_WATERMARK`), reading bytes from TCP sockets is paused for some
connections, allowing only a few to proceed until the pressure is resolved.
When memory grows too high (`HIGH_WATERMARK`) bulk items are rejected with 429.
This mechanism protects against unbounded memory usage and `OutOfMemory`
errors (OOMs).

ES supports multiple `Content-Type`s for the payload. These are
implementations of `MediaType` interface. A common implementation is called
`XContentType`, including CBOR, JSON, SMILE, YAML, and their versioned types.
X-pack extensions includes PLAIN_TEXT, CSV, etc. Classes that implement
`ToXContent` and friends can be serialized and sent over HTTP.

HTTP routing is based on a combination of Method and URI. For example,
`RestCreateIndexAction` handler uses `("PUT", "/{index}")`, where curly braces
indicate path variables. `RestBulkAction` specifies a list of routes

```java
@Override
  public List<Route> routes() {
    return List.of(
      new Route(POST, "/_bulk"),
      new Route(PUT, "/_bulk"),
      new Route(POST, "/{index}/_bulk"),
      new Route(PUT, "/{index}/_bulk")
    );
  }
```

Every REST handler must be declared in the `ActionModule` class in the
`initRestHandlers` method. Plugins implementing `ActionPlugin` can extend the
list of handlers via the `getRestHandlers` override. Every REST handler
should extend `BaseRestHandler`.

The REST handler’s job is to parse and validate the HTTP request and construct a
typed version of the request, often a Transport request. When security is
enabled, the HTTP layer handles authentication (based on headers), and the
Transport layer handles authorization.

Request handling flow from Java classes view goes as:

```
(if security enabled) Security.getHttpServerTransportWithHeadersValidator
-> `Netty4HttpServerTransport`
-> `AbstractHttpServerTransport`
-> `RestController`
-> `BaseRestHandler`
-> `Rest{Some}Action`
```

`Netty4HttpServerTransport` is a single implementation of
`AbstractHttpServerTransport` from the `transport-netty4`
module. The `x-pack/security` module injects TLS and headers validator.

## Transport

Transport is the term for node-to-node communication, utilizing a TCP-based
custom binary protocol. Every node acts as both a client and a server.
Node-to-node communication almost never uses HTTP transport (except for
reindex-from-remote).

`Netty4Transport` is the sole implementation of TCP transport, initializing
both the Transport client and server. The `x-pack/security` plugin provides
a secure version: `SecurityNetty4Transport` (with TLS and authentication).

A `Connection` between nodes is a pool of `Channel`s, where each channel is a
non-blocking TCP connection (Java NIO terminology). Once a cluster is
discovered, a `Connection` (pool of `Channel`s) is opened to every other node,
and every other node opens a `Connection` back. This results in two
`Connection`s between any two nodes `(A→B and B→A)`. A node sends requests only
on the `Connection` it opens (acting as a client). The default pool is around 13
`Channel`s, divided into sub-pools for different purposes (e.g., ping,
node-state, bulks). The pool structure is defined in the `ConnectionProfile`
class.

ES never behaves incorrectly (e.g. loses data) in the face of network outages
but it may become unavailable unless the network is stable. Network stability
between nodes is assumed, though connectivity issues remain a constant
challenge.

Request timeouts are discouraged, as Transport requests are guaranteed to
eventually receive a response, even without a timeout. `SO_KEEPALIVE` helps
detect and tear down dead connections. When a connection closes with an error,
the entire pool is closed, outstanding requests fail, and the pool is
reconnected.

There are no retries on the Transport layer itself. The application layer
decides when and how to retry (e.g., via `RetryableAction` or
`TransportMasterNodeAction`). In the future Transport framework might support
retries #95100.

Transport can multiplex requests and responses in a single `Channel`, but
cannot multiplex parts of messages. Each transport message must be fully
dispatched before the next can be sent. Proper application-layer sizing/chunking
of messages is recommended to ensure fairness of delivery across multiple
senders. A Transport message cannot be larger than 30% of heap (
`org.elasticsearch.transport.TcpTransport#THIRTY_PER_HEAP_SIZE`) or 2GB (due to
`org.elasticsearch.transport.Header#networkMessageSize` being an `int`).

The `TransportMessage` family tree includes various types (node-to-node,
broadcast, master node acknowledged) to ensure correct dispatch and response
handling. For example when a message must be accepted on all nodes.

## Other networking stacks

Snapshotting to remote repositories involves different networking clients
and SDKs. For example AWS SDK comes with Apache or Netty HTTP client, Azure
with Netty-based Project-Reactor, GCP uses default Java HTTP client.
Underlying clients may be reused between repositories, with varying levels of
control over networking settings.

There are other features such as SAML/JWT metadata reloading, Watcher HTTP
action, reindex and ML related features such as inference that also use HTTP
clients.

## Sync/Async IO and threading

ES handles a mix of I/O operations (disk, HTTP server,
Transport client/server, repositories), resulting in a combination of
synchronous and asynchronous styles. Asynchronous IO utilizes a small set of
threads by running small tasks, minimizing context switch. Synchronous IO
uses many threads and relies on an OS scheduler. ES typically runs with 100+
threads, where Async and Sync threads compete for resources.

## Netty

Netty is a networking framework/toolkit used extensively for HTTP and Transport
networks, providing foundational building blocks for networking applications.

### Event-Loop (Transport-Thread)

Netty is an Async IO framework, it runs with a few threads. An event-loop is
a thread that processes events for one or many `Channels` (TCP connections).
Every `Channel` has exactly one, unchanging event-loop, eliminating the need to
synchronize events within that `Channel`. A single, CPU-bound `Transport
ThreadPool` (e.g.,4 threads for 4 cores) serves all HTTP and Transport
servers and clients, handling potentially hundreds or thousands of connections.

Event-loop threads serve many connections each, it's critical to not block
threads for a long time. Fork any blocking operation or heavy computation to
another thread pool. Forking, however, comes with overhead. Do not fork
simple requests that can be served from memory and do not require heavy
computations (milliseconds).

Transport threads are monitored by `ThreadWatchdog`. A warning log appears if a
single task runs longer than 5 seconds. Slowness can be caused by blocking, GC
pauses, or CPU starvation from other thread pools.

### ByteBuf - byte buffers and reference counting

Netty's controlled memory allocation provides a performance edge by managing and
reusing byte buffer pools (e.g., pools of 1MiB byte chunks sliced into 16KiB
pages). Some pages might not be in use while taking up heap space and show up in
the heap dump.

Netty reads socket bytes into direct buffers, and ES copies them into pooled
byte-buffers (`CopyBytesSocketChannel`). The application is responsible for
retaining (increasing ref-count) and releasing (decreasing ref-count) for
pooled buffers.

Reference counting introduces two primary problems:

1. Use after release (free): Accessing a buffer after it has been explicitly
   released.
2. Never release (leak): Failing to release a buffer, leading to memory leaks.

The compiler does not help detect these issues. They require careful testing
using Netty's LeakDetector with a Paranoid level. It's enabled by default in
all tests.

### Async methods return futures

Every asynchronous operation in Netty returns a future. It is easy to forget
to check the result, as a following call always succeeds:

```java
ctx.write(message)
```

Check the result of an async operation:

```java
ctx.write(message).addListener(f -> { if (f.isSuccess() ...)});
```

### Chunk Encoding

#### XContent

### Wire Serialization

The transport binary format is Elasticsearch's other serialization system, entirely separate from XContent. It is the
node-to-node wire protocol (see [Transport](#transport)) and it does not use Java's built-in `Serializable`. The core
interface is
[`Writeable`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/common/io/stream/Writeable.java),
whose sole method is `void writeTo(StreamOutput out)`. By convention a `Writeable` also provides a symmetric
*deserialization constructor* taking a `StreamInput`, which keeps the type immutable and puts read and write logic side by
side:

```java
public MyClass(StreamInput in) throws IOException {
    this.count = in.readVInt();
    this.items = in.readCollectionAsList(Item::new);
}
```

[`StreamOutput`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/common/io/stream/StreamOutput.java)
and [`StreamInput`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/common/io/stream/StreamInput.java)
are abstract `java.io.OutputStream` / `InputStream` subclasses that form the write and read sides of this layer. They are
*typed* streams that know how to encode ES's primitive and common types (`writeVInt`, `writeString`, collection helpers,
etc.), and, critically, each carries the negotiated `TransportVersion` for the connection, so `writeTo` implementations
and deserialization constructors can branch for backwards compatibility. The current BWC idiom gates a field on a named
transport-version feature:

```java
// writing
if (out.getTransportVersion().supports(MY_FEATURE)) {
    out.writeString(newField);
}
// reading
this.newField = in.getTransportVersion().supports(MY_FEATURE) ? in.readString() : null;
```

where `MY_FEATURE` is a `public static final TransportVersion MY_FEATURE = TransportVersion.fromName("my_feature");`.
Prefer `supports(...)` over the older `onOrAfter(TransportVersions.X)` comparison. See
[Versioning.md](./Versioning.md) for the full transport-version workflow (how to declare a new feature, confirm backport
branches, and regenerate the resource files with `./gradlew generateTransportVersion`).

When the reader does not know the concrete type at compile time (for example a `QueryBuilder` that could be a
`TermQueryBuilder` or a `BoolQueryBuilder`), the type implements `NamedWriteable`. The name is written to the stream, a
`NamedWriteableRegistry` maps `(categoryClass, name)` to the concrete reader, and a `NamedWriteableAwareStreamInput` wraps
a `StreamInput` with that registry so named reads resolve transparently.

Serialization is exercised by dedicated test base classes: `AbstractWireSerializingTestCase<T>` (transport round-trip
through `StreamInput`/`StreamOutput`), `AbstractXContentSerializingTestCase<T>` (XContent parse/serialize round-trip), and
`AbstractBWCSerializationTestCase<T>` (round-trips across transport versions).

### Performance

(long running actions should be forked off of the Netty thread. Keep short operations to avoid forking costs)

### Work Queues

### RestClient

The `RestClient` is primarily used in testing, to send requests against cluster nodes in the same format as would users. There
are some uses of `RestClient`, via `RestClientBuilder`, in the production code. For example, remote reindex leverages the
`RestClient` internally as the REST client to the remote elasticsearch cluster, and to take advantage of the compatibility of
`RestClient` requests with much older elasticsearch versions. The `RestClient` is also used externally by the `Java API Client`
to communicate with Elasticsearch.

# Concurrency

Elasticsearch runs a mix of network IO, disk IO, and CPU-bound work across a small number of
network threads and a much larger number of worker threads. The rest of this chapter answers the
questions that come up when writing concurrent code in this codebase, in the order they naturally
arise. It starts with what must never block, then covers how work gets represented as a unit that
can carry its own error handling, where that unit of work actually runs, how its result or failure
gets back to the caller, and how shared state gets protected when several of these units run at
once.

## The core rule

The Netty event-loop threads described under [Event-Loop (Transport-Thread)](#event-loop-transport-thread)
serve many connections each, so blocking one for any real length of time stalls every other
connection on that thread. The rule that follows from this is simple to state and easy to violate.
Never block a transport or event-loop thread. Fork any blocking operation or heavy computation onto
a dedicated pool instead, and only fork work that is actually heavy, since forking has its own
overhead.

## Representing a unit of work

Submitting a plain `Runnable` via `execute(...)` works, but the task gets none of the structured handling ES relies on.
There is no built-in way to distinguish and handle *rejection* (when the queue is full) separately from a normal run exception.
Any exception thrown from `run()` propagates only as an uncaught exception on the worker thread rather than a callback tied to that specific task.
Calling `submit(...)` instead wraps the task in a `Future`, and the resulting exception is captured there silently unless
something later calls `get()` on it. ES therefore prefers two richer abstractions:

- [`AbstractRunnable`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/common/util/concurrent/AbstractRunnable.java)
  extends `Runnable` with structured exception handling and lifecycle hooks: `doRun()` holds the work, `onFailure(Exception)`
  receives any exception thrown by `doRun()`, `onRejection(Exception)` fires when the queue is full (defaulting to
  `onFailure`), `onAfter()` always runs like a `finally`, and `isForceExecution()` lets a task bypass a bounded queue's
  capacity limit.
- [`ActionRunnable`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/ActionRunnable.java)
  extends `AbstractRunnable` and bridges it to an `ActionListener`, so `onFailure` is routed to `listener.onFailure`.

## Choosing where it runs

The [`ThreadPool`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/threadpool/ThreadPool.java) class is the central registry for the node's shared named executors and the entry point for obtaining one. Rather than holding
references to raw `ExecutorService`s, code asks the `ThreadPool` for a named executor and submits work to it.

Elasticsearch does not spin up ad-hoc threads or executors. Every `Runnable` runs on a pool obtained
from `ThreadPool`. The main exception is third-party libraries such as the AWS, GCP, and Azure SDKs
used for snapshot repositories, which manage their own internal thread pools (see
[Other networking stacks](#other-networking-stacks)).

There are many named pools rather than one shared one because each workload needs its own resource
isolation. A saturated pool for one workload cannot starve another, and each pool gets its own
sizing, its own back-pressure policy, and its own monitoring through `_cat/thread_pool` and node
stats. A pool is more than just a named executor.

### Two pool types

Each pool has a [`ThreadPoolType`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/threadpool/ThreadPool.java#L159):

- A **FIXED** pool has a static number of threads, backed by a *bounded* or *unbounded* queue. When the queue fills up, new tasks are
  *rejected immediately* (surfacing as an `EsRejectedExecutionException`). This gives predictable resource consumption and
  fast back-pressure under load, so it is used for the hot data paths such as `SEARCH` and `WRITE`.
- A **SCALING** pool has a thread count that grows on demand between a min and a max, backed by an *unbounded* queue, and
  its idle threads expire after a keep-alive period. This suits bursty, infrequent work such as `GENERIC`, `MANAGEMENT`,
  and snapshots.

### Built-in pools

The built-in pools are declared as string constants in
[`ThreadPool.Names`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/threadpool/ThreadPool.java#L77).
Plugins register additional pools at startup by overriding
[`Plugin::getExecutorBuilders(Settings)`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/plugins/Plugin.java#L304).

`CLUSTER_COORDINATION` handles master elections, cluster membership updates, and sending cluster-state publication
requests. It is sized to a single thread to avoid contention on [Coordinator]'s internal mutex (see the
[Cluster Coordination](#cluster-coordination) section).
`GENERIC` scales up to
[somewhere between 128 and 512 threads](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/threadpool/DefaultBuiltInExecutorBuilders.java#L33)
depending on the number of processors, and is used throughout the codebase.
`WRITE` is a **FIXED** pool with
[`allocatedProcessors` threads and a large bounded queue](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/threadpool/DefaultBuiltInExecutorBuilders.java#L52),
sized to `max(allocatedProcessors * 750, 10000)`, and it tracks execution time and utilization for
autoscaling. It performs blocking disk IO through Lucene, which is exactly why it gets its own
dedicated bounded pool rather than sharing `GENERIC`. This is the same blocking-versus-non-blocking
placement rule from [The core rule](#the-core-rule) applied to a specific workload. The Netty
transport-worker threads that receive the request must never block on that IO, so the work is
handed off to another thread pool such as `WRITE` instead.

## How results and errors flow back

See the [Javadocs for `ActionListener`](https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/action/ActionListener.java)

`ActionListener<T>` is the primary async callback interface in Elasticsearch, a form of *continuation-passing style*.
Rather than blocking the caller until an asynchronous operation returns a value, the operation is instead handed "what
to do next" (the listener) as an argument. The listener is invoked once the operation result is ready. This allows for operations to be composed
without the caller needing to block and wait for completion. The contract is a
two-path pair of methods:

```java
void onResponse(T response); // success path
void onFailure(Exception e);  // failure path
```

The `ActionListener` Javadoc explains why ES deliberately avoids `java.util.concurrent.CompletableFuture` and similar
mechanisms in production code. They can achieve the same goals, but they permit *blocking* while waiting for a result,
almost never appropriate where threads are a precious resource, and they can *catch an `Error`*, delaying (or preventing)
the JVM exit that should follow such a fatal condition. `ActionListener` makes those misuses impossible.

### ThreadContext propagation

Every thread managed by ES carries a
[`ThreadContext`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java),
which is a map of string headers and transient objects attached to the current logical request, such as authentication
tokens, tracing IDs, custom request headers, the security context, and other transient objects that should flow with the
work but not be serialized over the wire. This is why an `ActionListener` invoked from a thread-pool
executor automatically carries the `ThreadContext` across that hop without any manual plumbing. The
[`EsThreadPoolExecutor`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/common/util/concurrent/EsThreadPoolExecutor.java#L174)
wraps every submitted task in a `ContextPreservingRunnable` at submission time, and the worker thread restores the
captured context before `doRun()` and cleans it up afterwards. As a result a request that begins with a user's auth
token and fans out across search, fetch, and coordination threads carries that token on every hop with no manual
threading of context through the code.

### Composition

`ActionListener` can be subclassed directly, usually as an anonymous class, when listeners
implement complex logic. It is important to note that an exception thrown from `onResponse` is not automatically routed to
`onFailure` and must be handled manually. `ActionListener` also
provides static factory methods for composing delegates without writing a full implementation. `wrap` builds an
inline listener from two lambdas, `runAfter` runs a cleanup action once the delegate completes, and
`delegateFailureAndWrap` handles only the success path while forwarding failures automatically, including failures
thrown by that success-path logic itself. See the
[`ActionListener`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/ActionListener.java)
class itself for the full set of factory methods and their exact semantics.

### Fan-out and specialized implementations

When one operation must wait for multiple concurrent sub-operations, there are purpose-built implementations:

- [`GroupedActionListener<T>`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/GroupedActionListener.java)
  handles a fixed, known number of parallel tasks and collects their results.
- [`RefCountingListener`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/RefCountingListener.java)
  handles a dynamic or unknown number of tasks, or cases where you only need to await all of them without collecting results.
- [`SubscribableListener<T>`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/SubscribableListener.java)
  lets multiple independent callers `addListener` to the same result so they are all notified on completion, and its `andThen`
  supports sequential CPS chaining of steps.
- [`ThreadedActionListener`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/ThreadedActionListener.java)
  wraps a delegate and dispatches its completion onto a specified `Executor`. This is the standard way to fork a
  listener's continuation off the Netty event-loop thread (see [The core rule](#the-core-rule) above and
  the [Performance](#performance) note below) so that expensive follow-up work does not run on an I/O thread.

### Pub/Sub Listeners

The `ActionListener` contract above is **one-shot**. A caller hands over a continuation, and that
continuation fires exactly once, for either the response or the failure. A different family of
listeners is **pub/sub** instead. A listener is registered once with a service and stays registered
for the lifetime of the node, or until explicitly deregistered, receiving a notification every time
a relevant event occurs. This is the classical Observer pattern, in which a subject maintains a list
of registered listeners and notifies all of them on each event. There is no notion of completion,
and the listener keeps receiving events. The two most common are
[`ClusterStateListener`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/ClusterStateListener.java)
(fires on every cluster-state update) and
[`IndexEventListener`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/shard/IndexEventListener.java)
(per-index and per-shard lifecycle events).

## Shared mutable state

This section is currently work-in-progress.

When multiple units of work run at once and need to share state, Elasticsearch leans on concurrent collections and the locks described in the
[Locking](#locking) chapter, but the stronger bias throughout the codebase is to avoid shared
mutable state in the first place. Cluster state is the clearest example. Rather than locking a
mutable structure, each update produces a new immutable `ClusterState` instance.

# Cluster Coordination

### Node Roles

[DiscoveryNodeRole]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNodeRole.java

[DiscoveryNode]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNode.java

Every Elasticsearch node in a cluster is assigned a set
of [roles](https://www.elastic.co/docs/deploy-manage/distributed-architecture/clusters-nodes-shards/node-roles) that
define the kind of responsibilities it can own. Roles are represented internally by the [DiscoveryNodeRole] class.
Possible values are `master`, `data`, `data_content`, `data_hot`, `data_warm`, `data_cold`, `data_frozen`, `ingest`,
`voting_only`, `remote_cluster_client`, `ml`, `transform`, `index` and `search`.

Roles are configured via the `node.roles` config field in `elasticsearch.yml`. If `node.roles` is not explicitly set,
the node starts with
all [default-enabled](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNodeRole.java#L72)
roles. If `node.roles` is explicitly set, only the listed roles are enabled.

A node with the `master` role is master-eligible: it can participate in master elections and can be elected as the
cluster's master node.

A node with the `data` role (relevant for stateful ES only) can host shards and perform data related operations such as
CRUD, search, and aggregations.

[Data-tier](https://www.elastic.co/docs/manage-data/lifecycle/data-tiers) roles (relevant for stateful ES only) allow
users to move indices to specific hardware specs as data grows older. This is managed
by [index lifecycle management policies](https://www.elastic.co/docs/manage-data/lifecycle/index-lifecycle-management).
The available data tier roles are:

- `data_content` for long-lived indices that can be queried and updated frequently, no matter how old they are.
- `data_hot`, which holds the most recent, and frequently queried, time-series data.
- `data_warm`, which typically holds less recent time-series data, that will not be read as often as the hot
  tier and rarely needs to be updated.
- `data_cold`, for infrequently accessed time-series data, sometimes in the form of searchable snapshots.
- `data_frozen`, which holds rarely accessed data stored entirely as searchable snapshots.

A node with the `ingest` role can
execute [ingest pipelines](https://www.elastic.co/docs/manage-data/ingest/transform-enrich/ingest-pipelines).

A node with the `ml` role can run machine learning jobs and trained model inference. Typically, those are memory and
CPU-intensive workloads so having dedicated nodes with this role avoids contention with indexing and searching.

A node with the `remote_cluster_client` role can establish outbound connections to remote clusters
for [cross-cluster search](https://www.elastic.co/docs/explore-analyze/cross-cluster-search) (CCS)
and [cross-cluster replication](https://www.elastic.co/docs/deploy-manage/tools/cross-cluster-replication) (CCR).

A node with the `transform` role can execute [transform](https://www.elastic.co/docs/explore-analyze/transforms) jobs.

A node with the `voting_only` role participates in master elections and cluster state publication quorum, but can never
be elected as master itself. The `voting_only`
role [must always be combined](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNodeRole.java#L229)
with the `master` role. Voting-only nodes are useful for providing an additional vote for quorum without requiring the
node to handle the full master
workload ([tiebreaker](https://www.elastic.co/docs/deploy-manage/distributed-architecture/clusters-nodes-shards/node-roles#voting-only-node)
nodes).

A node configured with an empty role list (`node.roles: []`) has no specific role and acts as
a [coordinating-only](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNode.java#L61)
node. Such nodes can receive client requests, route them to the appropriate data or master nodes, and aggregate results.
They do not hold data, run ingest pipelines, or participate in master elections.

#### Stateless Roles

In [stateless](https://www.elastic.co/blog/elastic-serverless-architecture) deployments, data is persisted in a shared
object storage and nodes' disk is only used as a cache layer. The `data` (and data tier) roles are disabled in favor of
two new roles:

- `index`: indexing nodes, which host primary shards and handle all write operations.
- `search`: search nodes, which host search-only replica shards and handle read operations.

It is worth distinguishing two terms that are often conflated. **Stateless** is the *architecture*. It separates storage
from compute, keeps durability in an object store, and treats nodes as ephemeral workers. **Serverless** is the managed
*offering* built on that architecture. It abstracts away provisioning, autoscaling, and error handling, billing users
for shared resource usage rather than for statically provisioned servers. The public
[Elastic serverless architecture blog](https://www.elastic.co/blog/elastic-serverless-architecture) contains an
architecture diagram of the two-tier design, and the design is described in detail in the SoCC'25 paper
[*Elasticsearch on the cloud: a stateless architecture*](https://dl.acm.org/doi/10.1145/3772052.3772245).

Splitting the write path (`index` nodes) from the read path (`search` nodes) is what lets read and write capacity scale
horizontally and independently. Stateless is implemented as an Elasticsearch plugin (the
[`stateless`](https://github.com/elastic/elasticsearch/tree/f607d700ce5e6d57d32e4e5f0f207de89704aeba/x-pack/plugin/stateless) plugin) that overrides the
relevant stateful components while reusing most of the stateful concepts described elsewhere in this guide.

#### Master Node Role

Master-eligible (`master` role) nodes form the cluster's [voting configuration](#quorum) and participate
in [master elections](#master-elections).

The elected master node is responsible for cluster-wide coordination: it processes all [cluster state](#cluster-state)
updates and publishes them to the rest of the cluster. The cluster maintains (conceptually, see [term](#term)
section) at most a single master at all times; if the master fails, a new one is elected from the remaining
master-eligible nodes (excluding `voting_only` nodes, which can vote but cannot be elected).

The next few sections explore the cluster state and the master election process in more detail.

### Cluster State

[ClusterState]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/ClusterState.java

[Metadata]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java

[PersistedClusterStateService]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/gateway/PersistedClusterStateService.java

[IndexMetadata]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/metadata/IndexMetadata.java

The [ClusterState] is the portion of the cluster state which is held in-memory by every node. To ensure
correctness, updates to the `ClusterState` require strong
consistency ([linearizability](https://en.wikipedia.org/wiki/Linearizability)). Without strong consistency, nodes could
observe conflicting cluster states (eg two nodes could believe they both hold the primary for the same shard, or apply
different mappings to the same index) leading to data loss or corruption.

The `ClusterState` is (conceptually) immutable: every update produces a new instance of the `ClusterState` class. The
elected master is responsible for coordinating all cluster state updates and publishing the latest to the other nodes in
the cluster. Note that `ClusterState` updates are expensive, often taking hundreds of milliseconds or more, so they
should only be executed when absolutely necessary.

The [Metadata] part of the `ClusterState` is persisted to disk via the [PersistedClusterStateService] and will survive
restarts. The rest of the `ClusterState` components are in-memory only, and need to be rebuilt every time there is a
full cluster restart.

#### Persisted State

[ClusterState]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/ClusterState.java

[Metadata]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java

[ProjectMetadata]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/metadata/ProjectMetadata.java

The [Metadata] of a [ClusterState] is persisted on disk and comprises information from two categories:

1. Cluster-scoped information

A few standouts are:

- `clusterUUID`: the cluster unique id.
- `coordinationMetadata`: the current term, voting configurations ... etc. (see [Master Elections](#master-elections)
  section).
- `persistentSettings`: cluster-level settings applied
  via [the cluster settings API](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-cluster-put-settings).
- `customs`: cluster-level custom metadata: `NodesShutdownMetadata`, `RepositoriesMetadata` .. etc.

2. Project-scoped information (located in the [ProjectMetadata])

Notable components of the `ProjectMetadata` include:

- `id`: the project unique id.
- `indices`: map of index name to [IndexMetadata] (settings, mappings, aliases, number of shards/replicas, etc.).
  Contains all indices in this project.
- `aliasedIndices` and `templates`. See [aliases](https://www.elastic.co/docs/manage-data/data-store/aliases)
  and [templates](https://www.elastic.co/docs/manage-data/data-store/templates) for more details.
- `customs`: project-level custom metadata. Includes data stream definitions (via `DataStreamMetadata` project custom),
  index lifecycle metadata (`IndexLifecycleMetadata`) and others.

Note that some concepts are applicable to both cluster and project scopes, e.g. [persistent tasks](#persistent-tasks).

#### Ephemeral State

[DiscoveryNodes]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNodes.java

[GlobalRoutingTable]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/routing/GlobalRoutingTable.java

[RoutingTable]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/routing/RoutingTable.java

[IndexRoutingTable]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/routing/IndexRoutingTable.java

[IndexShardRoutingTable]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/routing/IndexShardRoutingTable.java

[ShardRouting]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/routing/ShardRouting.java

[ClusterBlocks]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/block/ClusterBlocks.java

[SnapshotsInProgress]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/SnapshotsInProgress.java

[RestoreInProgress]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/RestoreInProgress.java

[SnapshotDeletionsInProgress]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/SnapshotDeletionsInProgress.java

[HealthMetadata]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/health/metadata/HealthMetadata.java

[CompatibilityVersions]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/version/CompatibilityVersions.java

[TransportVersion]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/TransportVersion.java

[ClusterFeatures]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/ClusterFeatures.java

The rest of the [ClusterState] is ephemeral, it is not persisted to disk and is rebuilt from scratch if the cluster
restarts. Some key components are:

- `nodes` ([DiscoveryNodes])

The set of all nodes currently in the cluster. Tracks the `masterNodeId`, the `localNodeId` and categorizes nodes by
role (master-eligible, data and ingest). Also records the minimum and maximum node versions and supported index versions
across the cluster.

- `routingTable` ([GlobalRoutingTable])

Maps each project to its [RoutingTable], which itself maps each index to an [IndexRoutingTable] containing
an [IndexShardRoutingTable] per shard. Each `IndexShardRoutingTable` lists the [ShardRouting] entries for a shard,
detailing which node owns its, its state (`UNASSIGNED`,`INITIALIZING`, `STARTED` or `RELOCATING`), and whether
it is a primary or replica.

- `blocks` ([ClusterBlocks])

Cluster-level, project-level and index-level blocks that restrict certain operations for the relevant scope. For
example, an index can be blocked for writes during a close operation, or the entire cluster can be set to read-only.
Blocks operate at
five [levels](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/block/ClusterBlockLevel.java):
`READ`, `WRITE`, `METADATA_READ`, `METADATA_WRITE`, and `REFRESH`.

- `customs`

Additional custom ephemeral state. Examples
include [SnapshotsInProgress], [RestoreInProgress], [SnapshotDeletionsInProgress], and [HealthMetadata].

- `compatibilityVersions` and `minVersions` ([CompatibilityVersions])

Per-node and cluster-wide min [TransportVersion] and system index mappings versions used to figure out serialization
compatibility across the cluster.

- `clusterFeatures` ([ClusterFeatures])

Info on what features are present throughout the cluster.

- Cluster state version fields

The `version` field is
a [monotonically increasing](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/ClusterState.java#L162)
`long` that uniquely identifies committed states. It is technically persisted to disk by [PersistedClusterStateService],
which stores it as `last_accepted_version` in the Lucene commit user data alongside the cluster metadata. On node
startup, this version is loaded back and used to initialize the in-memory cluster state, so the monotonically-increasing
property is preserved across restarts.
The `stateUUID` field uniquely identifies every state, including uncommitted ones. It is not persisted.

#### Master Service

[MasterService]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/service/MasterService.java

[Priority]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/common/Priority.java

[SnapshotsService]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java

[ClusterStateTaskExecutor]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/ClusterStateTaskExecutor.java

The [MasterService] is the single-threaded coordinator for all cluster state updates on the master node. It organizes
pending cluster state update tasks into [Priority]-based queues: `IMMEDIATE`, `URGENT`, `HIGH`, `NORMAL`, `LOW`, and
`LANGUID`. It processes them in that order, highest priority first.

The `MasterService` uses a batching framework that groups multiple cluster state update tasks together, processing them
as a single batch and publishing only one resulting `ClusterState` update. This avoids triggering a distinct publication
for each individual task, which would be very costly.

Producers of cluster state update tasks, such as [SnapshotsService], can
then [define](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/service/MasterService.java#L1516)
their own task queues, priority and batch executors ([ClusterStateTaskExecutor]), which the `MasterService` uses to
group and process related tasks together.

A [queue processor](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/service/MasterService.java#L1313),
backed by a single-threaded `java.util.concurrent.ExecutorService`, processes batches one at a time.
The [executeAndPublishBatch](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/service/MasterService.java#L336)
method takes the next batch, invokes the batch's `ClusterStateTaskExecutor` to compute a new [ClusterState], and
publishes the result.

If the new state is identical to the previous one (by reference), no publication takes place. Otherwise, the master
assigns a new `version` and `stateUUID` to the state and proceeds to publication.

#### Cluster State Publication

```mermaid
sequenceDiagram
    participant M as Master Node
    participant F as Follower Node

    rect rgb(255, 248, 240)
    Note over M,F: Phase 1 — Publish
    M->>F: PublishRequest (diff or full state)
    M->>M: Local PublishRequest/Response
    Note right of F: PublicationTransportHandler<br/>::handleIncomingPublishRequest<br/>Deserializes state
    Note right of F: CoordinationState<br/>::handlePublishRequest<br/>Validates term/version,<br/>persists as lastAcceptedState
    F->>M: PublishResponse (term, version)
    Note left of M: CoordinationState<br/>::handlePublishResponse<br/>Collects responses until quorum
    end

    rect rgb(255, 220, 220)
    alt Quorum not reached
        Note over M: FailedToCommitClusterStateException<br/>Master steps down, new election begins
    end
    end

    rect rgb(255, 248, 240)
    Note over M,F: Phase 2 — Commit
    M->>F: ApplyCommitRequest
    M->>M: Local ApplyCommitRequest/Response
    Note right of F: Coordinator::handleApplyCommit<br/>Marks last accepted state as committed
    Note right of F: ClusterApplierService<br/>::onNewClusterState<br/>Applies new ClusterState
    F->>M: ACK (after application completes)
    Note left of M: ClusterApplierService<br/>::onNewClusterState<br/>Master applies its own state
    end
```

[Coordinator]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java

[Publication]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Publication.java

[PublicationTransportHandler]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/PublicationTransportHandler.java

[CoordinationState]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationState.java

[ClusterStatePublisher]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/ClusterStatePublisher.java

[ApplyCommitRequest]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/ApplyCommitRequest.java

[ClusterStateDiff]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/ClusterState.java

[PublishRequest]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/PublishRequest.java#L19

[CoordinatorPublication]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L1944

[PublishResponse]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/PublishResponse.java#L21

Once the [MasterService] has computed a new [ClusterState], it passes it to the [Coordinator], which is responsible for
then sharing with it with all nodes in the cluster.
This [publication](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L1620)
process follows a two-step commit protocol.
The progress of the publication for each follower node is tracked via
a [PublicationTarget](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Publication.java#L235)
object which maintains a
[PublicationTargetState](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Publication.java#L226)
state.

The `Coordinator` creates
a [PublicationContext](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/PublicationTransportHandler.java#L261)
via the [PublicationTransportHandler], which pre-serializes the cluster state into shared, reference-counted byte
buffers. A [ClusterStateDiff] against the previous state is prepared. A full serialization is also prepared when
the previous state's `blocks` had its disable state persistence flag set, or when a node present in the new state was
absent from the previous one (since that node will not have a previous state to apply a diff against).
It then builds a [PublishRequest] and creates a [CoordinatorPublication]
to [coordinate](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L1711)
the protocol.

1. **Publish**

The master
then [sends](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/PublicationTransportHandler.java#L374)
the serialized cluster state to every node in the cluster. As an optimization, the diff is sent to nodes that were
already part of the cluster, while new nodes receive the full state directly. If a node
responds
with [IncompatibleClusterStateVersionException](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/IncompatibleClusterStateVersionException.java#L20)
the `PublicationTransportHandler`
then [retries](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/PublicationTransportHandler.java#L484)
with the full state.

The recipient
nodes [verify](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/PublicationTransportHandler.java#L125)
the received state is valid, compare it against their local state
via [CoordinationState.handlePublishRequest](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationState.java#L377)
and send back a [PublishResponse] containing the new cluster state term and version. At that point, the recipient nodes
have [persisted](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationState.java#L405)
the new state but have not yet applied it.

2. **Commit**

Once the master has collected enough `PublishResponse`s from master-eligible nodes to satisfy quorum (
see [Quorum](#quorum) section), it creates an [ApplyCommitRequest]. The master then sends this commit message to all
nodes that responded to the `PublishRequest`.

When receiving this request, each
node [marks this last accepted state as committed](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationState.java#L517)
and proceeds to apply the new state (see [Cluster State Application](#cluster-state-application) section). The node
only ACKs the `ApplyCommitRequest` back to the master after application completes (i.e. after
`ClusterApplierService.onNewClusterState` finishes).

If the master cannot achieve a [PublishResponse] quorum (e.g. too many nodes are faulty), the [Publication] will fail
the entire publication with
a [FailedToCommitClusterStateException](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/FailedToCommitClusterStateException.java).
The master will then step down (it can no longer be master if it is not able to publish cluster states) and a new
election begins.

Once the `ApplyCommitRequest`s have been sent, the master will
try [waiting](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Publication.java#L103)
for all nodes to apply the new state
before [applying its own state](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L2056)
and moving on. No quorum of responses is needed at that point. If follower nodes time out, the master will still
move on to applying the new cluster state locally.

#### Cluster State Application

[ClusterApplierService]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java

[ClusterStateApplier]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/ClusterStateApplier.java

[ClusterStateListener]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/ClusterStateListener.java

[ClusterChangedEvent]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/ClusterChangedEvent.java

When a new [ClusterState] is committed, the follower nodes need to apply the committed state locally.
This is the responsibility of the [ClusterApplierService] class, which runs on
a [single dedicated thread](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java#L159).

To save bandwidth, the master normally sends a follower only a *diff* against the state it believes that follower last
received (see the publication-side diff discussion above). The follower can reconstruct the complete `ClusterState`
from that diff only if its own locally cached `lastSeenClusterState` matches the diff's base state. A mismatch can
happen if the node just joined, restarted, or missed an intervening publication. When that happens, applying the diff
throws an [`IncompatibleClusterStateVersionException`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/IncompatibleClusterStateVersionException.java#L20)
and the master transparently resends that node the full state instead (see the Publish phase of
[Cluster State Publication](#cluster-state-publication) above). Either way, the
follower ends up with the complete state, and that reconstructed state is what gets exposed as the current state to the
appliers and listeners below.

When [receiving](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L412)
an [ApplyCommitRequest] from the master, the [Coordinator] will hand over the committed state to
the `ClusterApplierService`, which
will [submit](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java#L374)
an [UpdateTask](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java#L399)
to the executor. If the new state differs from the currently applied state (by reference), the
service [applies](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java#L538)
the changes through the following sequence of steps:

1. A [ClusterChangedEvent] is created, holding the new state, the previous state, and their deltas.

2. The
   node [opens](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java#L551)
   network connections to newly added nodes (blocking call).

3. If needed, the updated cluster settings
   are [applied](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java#L559).

4. [ClusterStateApplier]s (which update the node's internal state) are
   invoked [in priority order](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java#L590).
   Example high-priority appliers are [IndicesClusterStateService], which creates, deletes and updates local shard
   copies, and [RepositoriesService], which manages snapshot repository lifecycle.

5. Transport connections to nodes that are no longer in the cluster are closed (non-blocking call).

6. `ClusterApplierService::state`
   is [updated](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java#L569)
   to the new state, making it visible to the rest of the classes on the node.

7. [ClusterStateListener]s
   are [notified](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java#L571).
   Unlike appliers, listeners run _after_ the state is visible. Examples include the [PersistentTasksNodeService] which
   reacts to changes in persistent tasks, and the [SnapshotShardsService] which watches for snapshot-related shard
   assignments.

Once application completes, the `Coordinator`'s
callback [onClusterStateApplied](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L463)
closes the election scheduler (if active) and stops peer finding, since the node is now part of a cluster with a
functioning master.

#### Persistence

[LucenePersistedState]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/gateway/GatewayMetaState.java#L533

[AsyncPersistedState]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/gateway/GatewayMetaState.java#L391

[PersistedState]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationState.java#L540

As mentioned in previous sections, only the [Metadata] portion of the [ClusterState] is persisted to disk (
see [Persisted State](#persisted-state)). The rest of the `ClusterState` (routing table, node membership, in-progress
snapshots ... etc.) is ephemeral and rebuilt from scratch after each full cluster restart.

The [PersistedClusterStateService] is responsible for storing the cluster metadata in a simple Lucene index in each of
the node's data paths, under the
`_state` [directory](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/gateway/PersistedClusterStateService.java#L172).
The metadata is split across several documents and is written incrementally where possible.

The `PersistedClusterStateService` uses 3 types of documents, each identified by a `type` string field: `global`,
`index` and `mapping`. Each document type stores its content in the `data` field as compressed SMILE-encoded XContent.
All documents are paged (default 1MB per page).

The `global` document serializes almost all the Metadata object: coordination metadata, persistent settings, customs,
all project-scoped data, except individual index metadata and mappings. The `index` documents (one document per index,
keyed by `index_uuid`) store all the [IndexMetadata]: settings, aliases, number of shards, etc., except for
the index mapping. The `mapping` documents (one document per mapping, keyed by `mapping_hash`) store index mappings.
Storing mappings separately is a deduplication optimization: multiple indices that share the same mapping (common
occurrence, eg in data streams) all reference the same mapping hash, so only one copy is stored.

During [incremental writes](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/gateway/PersistedClusterStateService.java#L1011),
the `PersistedClusterStateService` will first add new mappings and delete obsolete ones. It will then compare the index
metadata for each index UUID between the old and new state. If the version changed, the old document is deleted and a
fresh one is written. Unchanged indices are skipped entirely. The `global` document update is not incremental. If the
new state's `global` fields differ from the old ones, the `global` document
is [deleted and rewritten entirely](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/gateway/PersistedClusterStateService.java#L1080).

Each Lucene commit
also [records](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/gateway/PersistedClusterStateService.java#L884)
the current term, the last accepted cluster state version, the node ID, the node version, the oldest index version (used
for compatibility checks on node startup) and the cluster UUID in
the [commit user data](https://lucene.apache.org/core/10_0_0/core/org/apache/lucene/index/IndexWriter.html#setLiveCommitData(java.lang.Iterable)).

The coordination layer interacts with the recorded state on disk through the [PersistedState] interface, which
the [CoordinationState] calls at three key points during the publication protocol:

- [setCurrentTerm](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationState.java#L195):
  when the node joins a new term (via `handleStartJoin`).
- [setLastAcceptedState](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationState.java#L405):
  after the node receives a [PublishRequest] from the master. The new state is persisted before
  the node sends its [PublishResponse].
- [markLastAcceptedStateAsCommitted](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationState.java#L517):
  when the node processes an [ApplyCommitRequest] from the master.

On master-eligible nodes, the `PersistedState` implementation is [LucenePersistedState], which writes synchronously on
the cluster coordination thread. On non-master-eligible nodes, the [AsyncPersistedState] wrapper is used instead. It
applies updates to an in-memory state immediately and queues the disk write to a background thread. This avoids blocking
the coordination thread on disk I/O for nodes that do not participate in master elections.

On startup, the
node [reads](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/gateway/PersistedClusterStateService.java#L432)
the Lucene index, selects the persisted state with the highest accepted term, and reconstructs the
`Metadata` from the stored documents and commit user data. This metadata is then used to build the initial
`ClusterState` that the node starts with.

### Master Elections

[Coordinator]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java

[CoordinationState]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationState.java

[Join]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Join.java

[JoinRequest]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/JoinRequest.java

[CoordinationMetadata]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationMetadata.java

[VotingConfiguration]: https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationMetadata.java#L326

[Reconfigurator]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Reconfigurator.java

The master's core responsibility is updating the [ClusterState] and propagating the changes to other nodes.

The cluster maintains (conceptually at least, see [term](#terms) section) at most a single master at all times. If no
master is
present, [master-eligible](https://www.elastic.co/docs/deploy-manage/distributed-architecture/clusters-nodes-shards/node-roles#master-node-role)
nodes will attempt to become the master by starting an election. The cluster will not be able to commit
any [ClusterState] changes until a new master is elected.

To elect a master, Elasticsearch uses a consensus algorithm derived
from [Paxos](https://lamport.azurewebsites.net/pubs/lamport-paxos.pdf) (see
also [Paxos Made Simple](https://lamport.azurewebsites.net/pubs/paxos-simple.pdf)
and [Raft](https://raft.github.io/) for more approachable resources). This algorithm is formally defined in a TLA+
specification referenced from the [CoordinationState] class. The [CoordinationState] class is responsible for
maintaining the [safety](https://en.wikipedia.org/wiki/Safety_and_liveness_properties#Safety) property of the algorithm,
whereas the [Coordinator] ensures [liveness](https://en.wikipedia.org/wiki/Safety_and_liveness_properties#Liveness),
by guaranteeing progress in cases of timeouts, failures and/or conflicts. The [Coordinator] also manages a node's
transition between `CANDIDATE`,`LEADER`, and
`FOLLOWER` [modes](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L1794),
and constructs the main data structures used for cluster state publication.

The [CoordinationMetadata] instance stored in [ClusterState]::[Metadata], tracks the coordination-related state needed
for subsequent master elections.

#### Quorum

A quorum is defined as a strict majority (`votedNodesCount * 2 > nodeIds.size()`) of the nodes listed in
a [VotingConfiguration]. [CoordinationMetadata] maintains two such configurations: `lastCommittedConfiguration` and
`lastAcceptedConfiguration`. Most of the time, these two configurations are equal. They only diverge during
[voting configuration changes](https://www.elastic.co/docs/deploy-manage/distributed-architecture/discovery-cluster-formation/modules-discovery-voting#_voting_configuration_updates).
The [Reconfigurator] handles the logic for computing the voting configuration changes during the transition.

To guarantee safety during voting configuration updates, an election will only succeed if quorum is achieved
for [both](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/ElectionStrategy.java#L61)
configurations.

#### Terms

Every election takes place in a
numbered [term](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationMetadata.java#L36).
Terms are monotonically increasing `long`s that act as logical clocks for the coordination layer, allowing the cluster
to distinguish between master mandates coming from distinct elections. They are persisted
via [CoordinationState.PersistedState](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationState.java#L42).
A term can have at most one master. A node will never vote in the same term twice.

Note that nodes may not necessarily be aware of the same latest term at the same physical time. The practical
consequence of this is that two nodes may temporarily both *believe* they are the master at the same physical time. But
because there is only ever a single master per term, terms are always monotonically increasing and cluster
state updates require a quorum of votes, at most one of them will be able to update the cluster state.

#### Election Flow

Below is an overview of the election process, including the main code components for each step. Subsequent sections
provide more detailed explanations of the mechanics and rationale behind each step.

1. Leader failure detected.

A follower detects current master failure.

```
LeaderChecker
Coordinator.onLeaderFailure()
```

2. Node becomes `CANDIDATE`.

The node transitions to `CANDIDATE` mode and triggers the discovery process.

```
Coordinator.becomeCandidate()
Mode.CANDIDATE
PeerFinder.activate(...)
```

3. Peer discovery

The candidate probes master-eligible nodes from the last accepted state and the configured hosts resolver. It asks if
there is a current master and what other master-eligible nodes are in the cluster.

```
PeerFinder.handleWakeUp()
ConfiguredHostsResolver
PeerFinder.startProbe(...)
PeersRequest
PeersResponse
```

4. Candidate schedules election.

If no current master is discovered and the node has enough master-eligible peers to form a quorum, the
candidate schedules an election (with randomized backoff to minimize conflicts with other potential
candidates).

```
CoordinatorPeerFinder.onFoundPeersUpdated()
Coordinator.startElectionScheduler()
ElectionScheduler.scheduleNextElection()
```

5. Pre-vote round.

The election starts. The candidate double checks it is healthy and eligible to be master. The candidate
sends pre-vote requests to master-eligible peers.

```
NodeEligibility
PreVoteCollector.start(...)
PreVoteRequest
PreVoteResponse
```

6. Pre-vote quorum reached

If enough `PreVoteResponse`s are received to satisfy quorum, the candidate proceeds to the real election.

```
ElectionStrategy.isElectionQuorum(...)
Coordinator.startElection()
```

7. Candidate bumps term and broadcasts `StartJoinRequest`.

The candidate computes the new term: `max(current, seen) + 1` and sends `StartJoinRequest` with the new term to all
nodes discovered in previous step.

```
Coordinator.getTermForNewElection()
Coordinator.broadcastStartJoinRequest(...)
StartJoinRequest
```

8. Other nodes start the join process

Master-eligible nodes ack the `StartJoinRequest`, persist the latest term, verifies they are healthy enough to cast a
vote, and send a `JoinRequest` to the candidate.

```
CoordinationState.handleStartJoin(...)
CoordinationState.setCurrentTerm(...)
JoinHelper.sendJoinRequest(...)
```

9. Candidate becomes `LEADER`

If enough `JoinRequest`s are received to satisfy quorum, the candidate becomes `LEADER`.

```
Coordinator.processJoinRequest(...)
CoordinationState.handleJoin(...)
ElectionStrategy.isElectionQuorum(...)
Coordinator.becomeLeader()
Mode.LEADER
```

10. Leader completes election

The leader publishes the cluster state, cleans up discovery connections, and starts heartbeating.

```
CandidateJoinAccumulator.close(...)
JoinTask
LeaderHeartbeatService.start(...)
```

#### Failure Detection

[LeaderChecker]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java

[FollowersChecker]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java

[FollowerChecker]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java#L274

[FollowerCheckRequest]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java#L437

[LeaderCheckRequest]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java#L399

The cluster uses two complementary checks to detect node failures:
the [LeaderChecker] (run by followers) and the [FollowersChecker] (run by the leader).

Peer-to-peer communication in Elasticsearch typically involves two distinct TCP connections, so having both checks is
necessary to ensure full connectivity. They also serve distinct conceptual purposes: when the leader becomes
unreachable, followers must detect it and start an election, whereas when a follower becomes unreachable, the leader
will remove it from the cluster so its shards can be reallocated.

Each follower runs a [LeaderChecker]
that [periodically sends](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java#L233)
a [LeaderCheckRequest] to the current master (every 1 second by default). Upon receiving the request, the
master [will verify](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java#L174)
that its local node is healthy, that it is still the elected master, and that the requesting
node is part of the current cluster state. If all conditions are met, it will
then [ack](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java#L120)
the request.

If the check fails more times than the configured limit (3 by default), if the master reported itself as unhealthy or if
the master disconnected, the follower will conclude that the
master [has failed](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java#L340).
The follower will stop the periodic check and notify the [Coordinator],
which will transition the node to `CANDIDATE` mode and start the election process.

On the master side, the [FollowersChecker] class maintains a map of nodes in the current cluster state
to [FollowerChecker] objects.

Each [FollowerChecker] periodically (by default every 1s) sends
a [FollowerCheckRequest] containing the current term and the identity of the master node. When receiving this request,
each follower will
first [verify](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java#L185)
that its local node is healthy (and throw if not), and then check whether the request term matches its current term.
If it does, then the follower
will [ack](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java#L199)
the request right away. Otherwise, if the request term is higher than the node current term, the request
is [handed off](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L382)
to the [Coordinator] so that the node can update its local state and become follower of the master in the new term.

If the follower disconnects or the check fails too many times (3 by default), the master will conclude that the node has
[failed](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java#L365)
and add it to
its [set](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/FollowersChecker.java#L102)
of `faultyNodes` and untrack its corresponding [FollowerChecker] object. The [Coordinator] will get notified and
eventually remove the node from the set of discovered nodes in the [ClusterState] (which will cause its shards to be
reallocated to other healthy nodes).

#### Discovery

[PeerFinder]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/discovery/PeerFinder.java

[PeersRequest]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/discovery/PeersRequest.java

[PeersResponse]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/PeersResponse.java

[SeedHostsProvider]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/discovery/SeedHostsProvider.java

[SeedHostsResolver]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/discovery/SeedHostsResolver.java

[SettingsBasedSeedHostsProvider]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/discovery/SettingsBasedSeedHostsProvider.java

[FileBasedSeedHostsProvider]: https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/discovery/FileBasedSeedHostsProvider.java

[AwsEc2SeedHostsProvider]: https://github.com/elastic/elasticsearch/blob/v9.3.0/plugins/discovery-ec2/src/main/java/org/elasticsearch/discovery/ec2/AwsEc2SeedHostsProvider.java

[DiscoveryPlugin]: https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/plugins/DiscoveryPlugin.java

Discovery is a fast "gossip-like" protocol by which a node in `CANDIDATE` mode locates master-eligible nodes in the
cluster and/or the master itself.
A node enters `CANDIDATE` mode either on startup or after detecting a master failure. The [PeerFinder] class contains
the core logic for the discovery process.

Discovery starts with both the nodes from
the [last accepted cluster state](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L963)
and a set of resolved seed addresses, that are obtained from the different [SeedHostsProvider] implementations:

- [SettingsBasedSeedHostsProvider]: reads addresses from the `discovery.seed_hosts` config. This config is read on
  startup and non-refreshable.
- [FileBasedSeedHostsProvider]: reads addresses from the runtime-refreshable `unicast_hosts.txt` file in the config
  directory. Note that because it's dynamic, using this file is usually preferred to the `discovery.seed_hosts` config.
- Additional seed providers registered by users via the [DiscoveryPlugin]. One example of such plugin is
  the [AwsEc2SeedHostsProvider], which uses the AWS EC2 client to find the relevant instances that could host
  elasticsearch nodes.

[SeedHostsResolver] is in charge of converting the seed hosts strings to transport addresses using DNS resolution.

When discovery
is [active](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/discovery/PeerFinder.java#L94), [PeerFinder]
will
run [handleWakeUp](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/discovery/PeerFinder.java#L296)
every 1s (by default). Each iteration will clean up disconnected peers, send a [PeersRequest] to known master-eligible
nodes (from the last accepted state and the resolved seed addresses)
and [schedule](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/discovery/PeerFinder.java#L319)
the next `handleWakeUp` iteration

When [receiving](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/discovery/PeerFinder.java#L534)
a [PeersResponse], [PeerFinder] will reach out to all peers specified in the response, including a potential master. If
the peer node reports itself as the master and its term is greater or equal to the local one, then the node will update
its local term
and [try to join](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L1817)
the master.

Because the node will eagerly reach out to every peer it learns about (both via the content of a [PeersResponse] or even
when receiving a new [PeersRequest]), the discovery protocol will find every other master-eligible node in at most
roughly `⌈log₂(D)+1⌉` steps, where D is the diameter of the graph of seed host configurations.

Note that discovery connections are managed separately from the regular cluster transport connections:
they are held by the [PeerFinder] and used only during discovery. Once a master is
found or elected, all discovery connections
are [released](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/discovery/PeerFinder.java#L166).

#### Pre-Vote

[PreVoteRequest]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/PreVoteRequest.java

[PreVoteResponse]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/PreVoteResponse.java

[ElectionSchedulerFactory]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/ElectionSchedulerFactory.java

The pre-vote phase takes place before a candidate bumps the term. It is meant to prevent a partitioned node from
disrupting a healthy term by forcing a re-election when it rejoins the cluster. It ensures that a candidate can only
proceed to broadcast a higher term once a quorum of peers agrees that there is no active leader.

When receiving a [PreVoteRequest] from a candidate node, the receiver
will [check](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/StatefulPreVoteCollector.java#L89)
whether it currently knows of an active leader. If not, and if the receiver is healthy, it will reply with
a [PreVoteResponse] containing its `currentTerm`, `lastAcceptedTerm`, and `lastAcceptedVersion`. If it still considers a
different node to be the leader, it will reject the request.

Pre-vote rounds are triggered by the [ElectionSchedulerFactory],
which schedules each attempt after a random delay that grows linearly with the number of failed attempts (up to a
configured [maximum](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/ElectionSchedulerFactory.java#L45)).
This randomized backoff reduces the chances that two candidates will run pre-vote rounds simultaneously, avoiding
conflicts and giving each attempt enough time to complete before the next one begins.

For additional details on the theory behind the pre-voting phase, see
the [Pre-voting in distributed consensus](https://davecturner.github.io/2017/08/17/paxos-pre-voting.html) article. This
[blog post](https://blog.cloudflare.com/a-byzantine-failure-in-the-real-world/) also describes a real-world example of a
liveness failure caused by this check being omitted.

#### Join

[JoinHelper]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/JoinHelper.java

[JoinTask]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/JoinTask.java

The Join process is the mechanism by which a node is added to the [ClusterState] by the master.
Most of the join-related logic is located in the [Coordinator] class, which still delegates specific components to
its [JoinHelper] [instance](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L153).

Within a master term, a node that starts up
will [send](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L1820)
a [JoinRequest] to the discovered master (see [Discovery](#discovery) section for more details) so it can get added
(back) to the cluster. The master
will [handle](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L651)
the incoming join request by
first [validating](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L678)
that it can connect back to the joining node, that the joining node is able to deserialize the
current [ClusterState], and that the request's term and
version [match](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationState.java#L253)
its local ones. Then, the master
will [trigger](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L1516)
voting reconfiguration if needed,
and [submit](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/JoinHelper.java#L479)
a [JoinTask] to the master queue
to [add](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/NodeJoinExecutor.java)
the new node to the [ClusterState].

During a master election, [JoinRequest]s from peer nodes also act as votes (via the
optional [Join](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/JoinRequest.java#L52)
field) for the candidate that started the election. The join requests are then processed slightly differently by the
candidate node. Rather than being submitted immediately to the master queue, incoming joins are buffered by
the [joinAccumulator](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/JoinHelper.java#L523).
Each [Join] vote is recorded in [CoordinationState]. Once quorum is reached, the candidate
will [become leader](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L889),
and [flush](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/JoinHelper.java#L545)
all accumulated joins into a single [JoinTask] for the master queue to process.

#### Serverless Elections

[AtomicRegisterPreVoteCollector]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/stateless/AtomicRegisterPreVoteCollector.java

[StoreHeartbeatService]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/stateless/StoreHeartbeatService.java

[HeartbeatStore]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/stateless/HeartbeatStore.java

[Heartbeat]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/stateless/Heartbeat.java

[SingleNodeReconfigurator]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/stateless/SingleNodeReconfigurator.java

Stateful and Serverless ES share the same [Coordinator] core logic, but they differ in how they check leader liveness,
and define quorum.

In Serverless, the elected master
periodically [writes](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/stateless/StoreHeartbeatService.java#L153)
a [Heartbeat] (containing the current term and timestamp) to an external blob store, the [HeartbeatStore], via
the [StoreHeartbeatService]. Other nodes check whether the leader is alive by reading the latest heartbeat from the
store.

In Serverless, the master uses a single node voting configuration with only itself (defined by
the [SingleNodeReconfigurator]). Committing and publishing a new cluster state therefore only requires the master's own
acknowledgment. Similarly, during elections, the pre-vote phase does not contact any peers. Instead,
the [AtomicRegisterPreVoteCollector]
solely [checks](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/stateless/AtomicRegisterPreVoteCollector.java#L35)
the latest heartbeat from the blob store and starts the election if it's older than the configured threshold (default 30
seconds). To become master, the candidate node will atomically override the current term in the blob store via a CAS
operation.

Beyond leadership, the lease/root blob also serves as a consistency gate during normal operation. Before a node acks a
translog upload it verifies that its local cluster state is consistent with the authoritative blob, via
[`StatelessClusterConsistencyService.ensureClusterStateConsistentWithRootBlob()`](https://github.com/elastic/elasticsearch/blob/f607d700ce5e6d57d32e4e5f0f207de89704aeba/x-pack/plugin/stateless/src/main/java/org/elasticsearch/xpack/stateless/cluster/coordination/StatelessClusterConsistencyService.java),
blocking writes if cluster membership has changed and the node's local state has not yet caught up. This prevents a node
that has fallen behind on cluster-state updates from durably acknowledging writes it should no longer be handling.

### New Cluster Formation

[ClusterBootstrapService]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/ClusterBootstrapService.java

[UnsafeBootstrapMasterCommand]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/UnsafeBootstrapMasterCommand.java

[DetachClusterCommand]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/DetachClusterCommand.java

When a cluster starts from scratch, no persisted cluster state exists on disk. The [PersistedClusterStateService]
returns an empty on-disk state with a `CoordinationMetadata` containing `VotingConfiguration.EMPTY_CONFIG` for both the
last-committed and last-accepted configurations. The cluster first needs to
[bootstrap](https://www.elastic.co/docs/deploy-manage/distributed-architecture/discovery-cluster-formation/modules-discovery-bootstrap-cluster)
and establish its initial voting configuration before an election can take place.
The [ClusterBootstrapService] is responsible for this one-time bootstrapping process and runs on master-eligible nodes.

For production clusters, the operator has to provide the list of master-eligible nodes that should form the initial
cluster via the [
`cluster.initial_master_nodes`](https://www.elastic.co/docs/deploy-manage/deploy/self-managed/important-settings-configuration#initial_master_nodes)
setting.
`ClusterBootstrapService` uses this setting
to [construct](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/ClusterBootstrapService.java#L107)
its `bootstrapRequirements`.

On startup, a master-eligible node will
first [become candidate](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L1198),
which will also start the discovery process (see [Discovery](#discovery) section). Each time the [PeerFinder] reports
newly discovered peers, `ClusterBootstrapService`
will [check](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/ClusterBootstrapService.java#L173)
whether the discovered nodes satisfy the bootstrap requirements, i.e. when
a [strict majority](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/ClusterBootstrapService.java#L198)
of the nodes specified in `cluster.initial_master_nodes` are found.

When this condition is met, `ClusterBootstrapService`
will [construct](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/ClusterBootstrapService.java#L265)
a [VotingConfiguration] from the discovered node IDs
and [set](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L1382)
the initial cluster state. The node will
then [schedule](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java#L1390)
an election (see [Master Elections](#master-elections) for more details). Any requirements that could not be matched to
a discovered node are added
as [placeholder](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/ClusterBootstrapService.java#L61)
entries. These placeholders occupy slots in the voting configuration's `nodeIds` set and affect the quorum size, but
they cannot cast a vote until the real nodes come online and replace them.

When `cluster.initial_master_nodes` is not present,
and [no discovery config is provided](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/ClusterBootstrapService.java#L122)
either, [ClusterBootstrapService]
will [schedule](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/ClusterBootstrapService.java#L204)
a best-effort bootstrap after the `discovery.unconfigured_bootstrap_timeout` (default 3 seconds). This simply uses all
master-eligible nodes discovered so far. This is inherently unsafe and is only intended for development and testing.

#### Single-Node Discovery

If `discovery.type: single-node` is set, [ClusterBootstrapService] uses the local node name as the sole bootstrap
requirement. The node bootstraps itself immediately.

#### Disaster Recovery

If more than half of the master-eligible nodes are permanently lost and no snapshot is available, the last-resort
[UnsafeBootstrapMasterCommand] (`elasticsearch-node unsafe-bootstrap`) can force a single surviving master-eligible
node to become the new cluster leader. It works by [setting](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/UnsafeBootstrapMasterCommand.java#L91) both voting configurations
to a single-node configuration containing only the local node and [regenerating](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/UnsafeBootstrapMasterCommand.java#L102) the cluster UUID.

The remaining data nodes cannot join the newly bootstrapped master because their persisted state still references the
old cluster. The [DetachClusterCommand] (`elasticsearch-node detach-cluster`) sets both their voting configurations
to `VotingConfiguration.MUST_JOIN_ELECTED_MASTER`, containing the single node ID `_must_join_elected_master_`,
which prevents nodes from starting their own elections and forces them to discover and join the already-elected master.
It then [resets](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/DetachClusterCommand.java#L57)
the persisted term to `0` and marks `clusterUUIDCommitted` as false.

### Master Transport Actions

[TransportMasterNodeAction]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/master/TransportMasterNodeAction.java

Many cluster operations
([creating indices](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/admin/indices/create/TransportCreateIndexAction.java),
[managing snapshots](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/create/TransportCreateSnapshotAction.java), etc.)
run on the elected master node because they result in [ClusterState] updates. The [TransportMasterNodeAction] class
is the base class for such operations. It provides a common framework that handles routing requests to the current
master, retrying when the master changes, and checking for cluster blocks.

See `TransportMasterNodeAction` Javadoc for a detailed description of the execution flow and retry mechanism.

# Replication

(More Topics: ReplicationTracker concepts / highlights.)

### What is a Shard

### Primary Shard Selection

(How a primary shard is chosen)

#### Versioning

(terms and such)

### How Data Replicates

(How an index write replicates across shards -- TransportReplicationAction?)

### Consistency Guarantees

(What guarantees do we give the user about persistence and readability?)

# Locking

(rarely use locks)

At the coarsest level, a node acquires a filesystem lock on the `node.lock` file in its data directory at startup (via
[NodeEnvironment]), which prevents two Elasticsearch processes from sharing the same data path. Finer-grained shard,
translog, and Lucene locks are described below (see also the [On-Disk Shard Layout](#on-disk-shard-layout) and
[Shard Locking Layers](#shard-locking-layers) discussions under [Engine & Store](#engine--store)).

### ShardLock

### Translog / Engine Locking

### Lucene Locking

# Engine & Store

[Engine]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/Engine.java

[InternalEngine]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

[IndexShard]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

[Translog]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/translog/Translog.java

### IndexShard

The [IndexShard] class is the single entry point for all shard-level operations: indexing, deletion, real-time GET,
refresh, flush, recovery, and snapshot. There is exactly one `IndexShard` object per allocated shard.

An `IndexShard` holds references to:

- The shard's [Store], which wraps a Lucene `Directory` and provides access to the shard's Lucene index files on disk.
- The shard's [Engine], which manages all indexing and search operations for this shard, writing to both the
  [Translog] and the Lucene files managed by the `Store`.

The lifecycle of these objects (creation, recovery, and teardown) is controlled by the [IndicesClusterStateService],
which reacts to cluster state changes and updates local state accordingly (see
the [IndicesClusterStateService](#indicesclusterstateservice) section).

### Store

[Store]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/store/Store.java

[RefCounted]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/core/RefCounted.java

[StoreFileMetadata]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/store/StoreFileMetadata.java

[FsDirectoryFactory]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/store/FsDirectoryFactory.java

The [Store] is the lowest-level Elasticsearch persistence abstraction for a shard. Each shard has a single
dedicated `Store` that wraps a
Lucene [Directory](https://lucene.apache.org/core/10_3_2/core/org/apache/lucene/store/Directory.html), which is Lucene's
own file-system abstraction used to read and write index files on disk.
Lucene's `Directory` is a pure I/O abstraction: callers open
an [IndexInput](https://lucene.apache.org/core/10_3_2/core/org/apache/lucene/store/IndexInput.html) to read a named file
and create an [IndexOutput](https://lucene.apache.org/core/10_3_2/core/org/apache/lucene/store/IndexOutput.html) to
write one. The `Store` builds on the Lucene `Directory` capabilities by adding reference counting and corruption
detection, exposing committed file metadata and enforcing integrity invariants.

#### Reference Counting and Lifecycle

The `Store` implements [RefCounted]. Callers call `store.incRef()` before using it and `store.decRef()` in
a `finally` block when done. Once the reference count drops to zero the store is closed and the underlying Lucene
directory is cleaned up. The `Store` also receives a [ShardLock] at construction time and only releases it
once closed, allowing other threads waiting to acquire the lock for this shard to proceed.

#### Backing Directory

The Lucene `Directory` used by a `Store` is created by
an [IndexStorePlugin.DirectoryFactory](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/plugins/IndexStorePlugin.java#L40).
The default built-in implementation is [FsDirectoryFactory], which
supports [several store types](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/store/FsDirectoryFactory.java#L107)
selectable via the
`index.store.type` [setting](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/IndexModule.java#L111):

- `hybridfs` (default):
  a [HybridDirectory](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/store/FsDirectoryFactory.java#L180)
  that delegates
  to a [MMapDirectory](https://lucene.apache.org/core/10_3_2/core/org/apache/lucene/store/MMapDirectory.html) for
  performance-sensitive file types (postings, term vectors index, norms, vectors, etc.)
  and [NIOFSDirectory](https://lucene.apache.org/core/10_3_2/core/org/apache/lucene/store/NIOFSDirectory.html) for files
  where sequential access is preferred (e.g. stored fields). Direct I/O is also enabled for vector index files when
  supported by the OS. The `HybridDirectory` selects between mmap and NIO on a per-file basis
  by [checking](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/store/FsDirectoryFactory.java#L252)
  the file's Lucene extension and I/O context.
- `mmapfs`: uses `MMapDirectory` for all files.
- `niofs`: uses `NIOFSDirectory` for all files.
- `fs`: automatically selects the best type for the underlying file system.

#### MetadataSnapshot

The `Store`
exposes [MetadataSnapshots](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/store/Store.java#L820)
for each index commit, read and constructed from the Lucene `segments_N` files. A `MetadataSnapshot` is a point-in-time
map from filename to [StoreFileMetadata] for the committed files belonging to a Lucene index commit, produced by an
Elasticsearch flush.
Each [StoreFileMetadata] includes the file's name, on-disk length, CRC32 checksum (from the Lucene file footer), the
Lucene version that wrote it, and a `writerUuid` that uniquely identifies the writer.

`MetadataSnapshot`s are leveraged by several Elasticsearch workflows, including [peer recovery](#peer-recovery) and
replica shard allocation (via `TransportNodesListShardStoreMetadata`). They are used to compare the on-disk state of two
distinct shards and calculate how much data needs to be transferred to bring them into sync.

#### On-Disk Shard Layout

The `Store` and `Translog` files for every shard live under a node's data path. The
directory tree below the configured `path.data` looks like:

```
data/
└── nodes/
    └── 0/
        ├── node.lock                       # node-level lock (see below)
        └── indices/
            └── <index-uuid>/               # one directory per index
                ├── _state/
                │   └── state-<N>.st         # index metadata (settings, mappings)
                └── <shard-id>/              # e.g. 0, 1, 2 ...
                    ├── _state/
                    │   └── state-<N>.st     # primary flag, index UUID, allocation ID
                    ├── index/               # Lucene segment files + write.lock
                    │   ├── segments_N
                    │   └── ...              # see the Lucene File Layout table below
                    └── translog/
                        ├── translog-<N>.tlog # translog generation files
                        └── translog.ckp      # translog checkpoint (min/max seqno)
```

The [`node.lock`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/env/NodeEnvironment.java#L186)
file is a filesystem lock acquired by [NodeEnvironment] at startup, and it prevents two Elasticsearch processes from
sharing the same data directory (see the [Locking](#locking) section). The `_state/state-<N>.st` files are written
via `MetadataStateFormat`. At the index level they hold the index metadata (settings, mappings). At the shard level
they hold a small [`ShardStateMetadata`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/shard/ShardStateMetadata.java)
record. That record contains only a primary/replica flag, the owning index's UUID, and the shard's `AllocationId`. The `-<N>` suffix is a monotonically
increasing generation so a new state can be written and fsynced before the old one is removed. The `index/` directory holds the Lucene segments and commit points (its contents are
detailed in the [Lucene File Layout](#lucene-file-layout) table). The `translog/` directory holds the durability log
described under [Translog](#translog), where `translog.ckp` is the checkpoint tracking the current generation and its
sequence-number range.

The runtime object graph that owns these files is a containment hierarchy rooted at the shard:
`IndexShard -> Engine (InternalEngine) -> { IndexWriter, DirectoryReader, Store -> Directory }`. The `IndexWriter`
writes documents and drives flushes and merges, the `DirectoryReader` opens segments for search, and the `Store`
wraps the Lucene `Directory` for raw file I/O.
[`InternalEngine#getIndexWriterConfig()`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java)
assembles the write-path customizations used to build the `IndexWriter`, including the index deletion policy, the
soft-deletes field, the wrapped `TieredMergePolicy`, and codec selection. Read-path customizations such as the query
cache and query caching policy are attached separately, when the `Searcher` is built from the `DirectoryReader`.

#### Shard Locking Layers

[NodeEnvironment]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/env/NodeEnvironment.java

[ShardLock]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/env/ShardLock.java

Access to a shard's on-disk data is protected by three layers of locking, each scoped to a different level.

The [ShardLock] is a node-wide, coarse-grained lock managed by [NodeEnvironment]. It is backed by a
`Semaphore` and guarantees that at most one owner at a time has write access to a given shard directory
within a JVM process. The
`Store` [is given](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/store/Store.java#L169)
a `ShardLock` at creation time. It holds this lock for its entire lifetime, ensuring that write operations (e.g.
creating an `IndexWriter`, deleting shard files, or recovering from another shard) have exclusive access to the shard
directory.
Callers that need to access the directory without a live `Store` (e.g. `TransportNodesListShardStoreMetadata` reading
metadata for allocation
decisions) [acquire a temporary](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/indices/store/TransportNodesListShardStoreMetadata.java#L182)
`ShardLock` for the duration of the read.

The [metadataLock](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/store/Store.java#L168)
is an in-process `ReentrantReadWriteLock` inside the `Store`. It guards access to the
directory's file listing and segment metadata. Operations that only read metadata (e.g. `getMetadata` with an
existing
commit) [take the read lock](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/store/Store.java#L300),
allowing concurrent readers. Operations that structurally modify the
directory (e.g. renaming temporary recovery files via `renameTempFilesSafe`, cleaning up stale files
via `cleanupAndVerify`, or running `CheckIndex`) take the write lock, which excludes all readers and other
writers until the operation completes.

Lucene's [IndexWriter](https://lucene.apache.org/core/10_3_2/core/org/apache/lucene/index/IndexWriter.html) write lock
is a native file lock (`write.lock`) that Lucene places in the index
directory. It prevents two `IndexWriter` instances from ever opening the same directory simultaneously, even
across processes. When the `Store` needs to read metadata from a directory that has no active engine (e.g.
during `getMetadata(commit=null, lockDirectory=true)`), it acquires the Lucene write lock together with the
`metadataLock` write lock, ensuring no writer can interfere. In normal operation, the running `InternalEngine`
already holds this
lock [through its IndexWriter](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java#L149).

To summarize, `ShardLock` is a JVM-level lock enforced across the entire node, `metadataLock`
coordinates in-process readers and writers within the `Store`, and the Lucene write lock guards the raw
directory at the file-system level.

#### Corruption

The `Store` maintains
corruption [markers](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/store/Store.java#L142)
as special files prefixed with `corrupted_` written into the Lucene directory.
When an unrecoverable I/O error or checksum mismatch
is [detected](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/store/Store.java#L1372),
a marker file is written containing the serialized exception. Subsequent calls to `failIfCorrupted()` scan for these
marker files and throw a `CorruptIndexException` if any are found, preventing any further operations on a known-bad
shard. Corruption markers are removed only after the shard has been successfully recovered from another source (e.g. a
primary or a snapshot).

### Engine

The [Engine] abstract class is the Elasticsearch abstraction that manages and coordinates operations on the running
shard index. Where the `Store` manages files on disk, the `Engine` owns the write
path (indexing, deletion, no-ops), the read path (searcher acquisition, real-time GET), and Lucene lifecycle
operations (refresh, flush, merge). It also controls the translog, ensuring that every acknowledged write is
durably recorded before a response is sent.

The main implementation is [InternalEngine], used for all read-write shards (primary and replica). Other
implementations serve more specialized roles. For example,
the [ReadOnlyEngine](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/ReadOnlyEngine.java)
gives read-only access to a frozen shard, and
the [NoOpEngine](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/NoOpEngine.java)
acts as a placeholder for shards belonging to a closed index. It exists to allow shards of closed indices to be
correctly replicated in case of a node failure.

#### Segments, Refresh, and Flush

Lucene organizes index data on disk into immutable files called *segments*. They are created whenever the in-memory
write buffer is flushed or when merges combine existing segments into larger ones.
The [IndexWriter](https://lucene.apache.org/core/10_3_2/core/org/apache/lucene/index/IndexWriter.html)
accumulates writes in memory and periodically flushes new segments. The `InternalEngine` wraps an `IndexWriter` (for
writes) and a pair of internal and
external [reader manager](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java#L151)
that wrap
Lucene [DirectoryReader](https://lucene.apache.org/core/10_3_2/core/org/apache/lucene/index/DirectoryReader.html)s and
map Elasticsearch concepts onto Lucene's:

- An Elasticsearch `refresh` triggers a Lucene NRT reader reopen (via `DirectoryReader.openIfChanged()`), making
  recently indexed documents searchable without writing a full commit to disk. Refreshes do not call
  `IndexWriter.commit()` and do not persist data durably.
- An Elasticsearch `flush` calls `IndexWriter.commit()`, writing a durable commit point to disk. A flush is
  what allows the translog to be safely truncated up to that commit.

This distinction is a core piece of Elasticsearch's durability model. Documents are searchable after a refresh but
only durably stored after a flush. In between, the translog bridges the gap. Every write is appended to the
translog on disk before being acknowledged, so that it can be replayed in the event of a crash.

The detailed mechanics of how a write flows through the engine (including translog interaction, sequence number
assignment, and version map updates) are covered in the [Translog](#translog) and [Indexing / CRUD](#indexing--crud)
sections.

### IndicesClusterStateService

[ClusterStateApplier]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/ClusterStateApplier.java

[ShardStateAction]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/action/shard/ShardStateAction.java

[RecoverySource]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/routing/RecoverySource.java

[ShardRouting]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/routing/ShardRouting.java

The [IndicesClusterStateService] is a high-priority [ClusterStateApplier] of the [ClusterApplierService] (
see [Cluster State Application](#cluster-state-application)).
Any time a new cluster state is published, [IndicesClusterStateService] checks whether the state of indices
and shards has changed and updates the local node's shards to match accordingly.

Its `doApplyClusterState` method goes through the following operations in order each time a new cluster state
is applied:

1. [Delete indices](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java#L353)
   that no longer exist in the new cluster state: closes each `IndexShard` (which closes its
   `Engine`, flushing first), releases the `Store` reference, and deletes the on-disk shard directory.
2. [Remove shards](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java#L355)
   that are no longer assigned to this node: closes the `Engine` (without flushing, unflushed data is already in the
   translog and safe), releases the `Store` reference, but leaves the on-disk files intact for later cleanup by
   `IndicesStore`.
3. [Update index metadata](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java#L357):
   if index settings changed, propagates them in-memory to each `IndexShard` and then to its `Engine` (e.g. merge
   scheduler config, GC deletes policy, soft-delete retention). If mappings changed, updates the `MapperService`. No
   disk writes happen here.
4. [Create or update shards](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java#L359):
   for each [ShardRouting] targeting this node in `INITIALIZING` state, creates a new `Store` and `IndexShard` and kicks
   off recovery. For already-active shards, updates their routing metadata in-place.

The below diagram illustrates the end-to-end flow from a master publishing a new cluster state containing a
newly assigned shard down to the `Engine` becoming active.

```mermaid
sequenceDiagram
    participant M as Master Node
    participant CA as ClusterApplierService
    participant ICSS as IndicesClusterStateService
    participant IS as IndexShard
    participant S as Store
    participant E as InternalEngine

    rect rgb(255, 248, 240)
    Note over M,CA: Cluster State Committed
    M->>CA: ApplyCommitRequest (new ClusterState)
    CA->>ICSS: applyClusterState(ClusterChangedEvent)
    end

    rect rgb(240, 248, 255)
    Note over ICSS,S: Shard Creation
    ICSS->>ICSS: doApplyClusterState()<br/>deleteIndices()<br/>removeIndicesAndShards()<br/>updateIndices()
    ICSS->>S: new Store(shardDirectory)
    ICSS->>IS: new IndexShard(store, shardRouting, ...)
    ICSS->>IS: startRecovery(recoverySource)
    end

    rect rgb(240, 255, 240)
    Note over IS,E: Recovery and Engine Start
    IS->>S: validate / prepare directory
    IS->>E: new InternalEngine(engineConfig)
    E->>S: IndexWriter.open(store.directory)
    Note right of E: Engine open, shard STARTED
    end

    rect rgb(255, 240, 240)
    Note over IS,M: Report Back to Master
    IS->>M: ShardStateAction.shardStarted()
    Note left of M: Master updates ClusterState<br/>marks shard as STARTED
    end
```

For a newly assigned shard, `createShard`
first [acquires](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java#L754)
a `ShardLock` (preventing concurrent access to the same shard directory from a concurrent request) or throws a
`ShardLockObtainFailedException` if it fails to do so. It
then [creates](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/IndexService.java#L560)
a `Store` referencing the shard's on-disk directory,
and [instantiates](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/IndexService.java#L569)
an `IndexShard`. Recovery is
then [triggered](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/indices/IndicesService.java#L993)
based on the [RecoverySource] in the [ShardRouting].

The details of recovery, including how the `Engine` is created and started, are covered in the
[Peer Recovery](#peer-recovery), [Snapshot Recovery](#snapshot-recovery), and [Local Shards Recovery](#local-shards-recovery)
sections.

### Translog

[Basic write model]:https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-replication.html
[`Translog`]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/index/translog/Translog.java
[`InternalEngine`]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

It is important to understand first the [Basic write model] of documents:
documents are written to Lucene in-memory buffers, then "refreshed" to searchable segments which may not be persisted on disk, and finally "flushed" to a durable Lucene commit on disk.
If this was the only way we stored the data, we would have to delay the response to every write request until after the data had been flushed to disk, which could take many seconds or longer. If we didn't, it would mean that we would lose newly ingested data if there was an outage between sending the response and flushing the data to disk.
For this reason, newly ingested data is also written to a shard's [`Translog`], whose main purpose is to persist uncommitted operations (e.g., document insertions or deletions), so they can be replayed by just reading them sequentially from the translog during [recovery](#recovery) in the event of ephemeral failures such as a crash or power loss.
The translog can persist operations quicker than a Lucene commit, because it just stores raw operations / documents without the analysis and indexing that Lucene does.
The translog is always persisted and fsync'ed on disk before acknowledging writes back to the user.
This can be seen in [`InternalEngine`] which calls the `add()` method of the translog to append operations, e.g., its `index()` method at some point adds a document insertion operation to the translog.
The translog ultimately truncates operations once they have been flushed to disk by a Lucene commit; indeed, in some sense the point of a "flush" is to clear out the translog.

Main usages of the translog are:

* During recovery, an index shard can be recovered up to at least the last acknowledged operation by replaying the translog onto the last flushed commit of the shard.
* Facilitate real-time (m)GETs of documents without refreshing.

#### Translog Truncation

[Flush API]:https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-flush.html
[`INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING`]:https://github.com/elastic/elasticsearch/blob/dd1db5031ee7fdac284753c0c3b096b0e981d71a/server/src/main/java/org/elasticsearch/index/IndexSettings.java#L352
[`INDEX_TRANSLOG_FLUSH_THRESHOLD_AGE_SETTING`]:https://github.com/elastic/elasticsearch/blob/dd1db5031ee7fdac284753c0c3b096b0e981d71a/server/src/main/java/org/elasticsearch/index/IndexSettings.java#L370

Translog files are automatically truncated when they are no longer needed, specifically after all their operations have been persisted by Lucene commits on disk.
Lucene commits are initiated by flushes (e.g., with the index [Flush API]).

Flushes may also be automatically initiated by Elasticsearch, e.g., if the translog exceeds a configurable size [`INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING`] or age [`INDEX_TRANSLOG_FLUSH_THRESHOLD_AGE_SETTING`], which ultimately truncates the translog as well.

#### Acknowledging writes

[`index()` or `delete()`]:https://github.com/elastic/elasticsearch/blob/591fa87e43a509d3eadfdbbb296cdf08453ea91a/server/src/main/java/org/elasticsearch/index/engine/Engine.java#L546-L564
[`TransportWriteAction`]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/action/support/replication/TransportWriteAction.java
[`indexShard.syncAfterWrite()`]:https://github.com/elastic/elasticsearch/blob/387eef070c25ed57e4139158e7e7e0ed097c8c98/server/src/main/java/org/elasticsearch/action/support/replication/TransportWriteAction.java#L548
[`Location`]:https://github.com/elastic/elasticsearch/blob/693f3bfe30271d77a6b3147e4519b4915cbb395d/server/src/main/java/org/elasticsearch/index/translog/Translog.java#L977
[`AsyncIOProcessor`]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/common/util/concurrent/AsyncIOProcessor.java

A bulk request will repeatedly call ultimately the Engine methods such as [`index()` or `delete()`] which adds
operations to the Translog.
Finally, the AfterWrite action of the [`TransportWriteAction`] will call [`indexShard.syncAfterWrite()`] which will put the last written translog [`Location`] of the bulk request into a [`AsyncIOProcessor`] that is responsible for gradually fsync'ing the Translog and notifying any waiters.
Ultimately the bulk request is notified that the translog has fsync'ed past the requested location, and can continue to acknowledge the bulk request.
This process involves multiple writes to the translog before the next fsync(), and this is done so that we amortize the cost of the translog's fsync() operations across all writes.

#### Translog internals

[`Checkpoint`]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/index/translog/Checkpoint.java
[`Location`]:https://github.com/elastic/elasticsearch/blob/693f3bfe30271d77a6b3147e4519b4915cbb395d/server/src/main/java/org/elasticsearch/index/translog/Translog.java#L977
[`Operation`]:https://github.com/elastic/elasticsearch/blob/693f3bfe30271d77a6b3147e4519b4915cbb395d/server/src/main/java/org/elasticsearch/index/translog/Translog.java#L1087
[`Snapshot`]:https://github.com/elastic/elasticsearch/blob/693f3bfe30271d77a6b3147e4519b4915cbb395d/server/src/main/java/org/elasticsearch/index/translog/Translog.java#L711
[`sync()`]:https://github.com/elastic/elasticsearch/blob/693f3bfe30271d77a6b3147e4519b4915cbb395d/server/src/main/java/org/elasticsearch/index/translog/Translog.java#L813
[`rollGeneration()`]:https://github.com/elastic/elasticsearch/blob/693f3bfe30271d77a6b3147e4519b4915cbb395d/server/src/main/java/org/elasticsearch/index/translog/Translog.java#L1656
[`createEmptyTranslog()`]:https://github.com/elastic/elasticsearch/blob/693f3bfe30271d77a6b3147e4519b4915cbb395d/server/src/main/java/org/elasticsearch/index/translog/Translog.java#L1929
[`TranslogHeader`]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/index/translog/TranslogHeader.java
[`TranslogReader`]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/index/translog/TranslogReader.java
[`TranslogSnapshot`]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/index/translog/TranslogSnapshot.java
[`MultiSnapshot`]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/index/translog/MultiSnapshot.java
[`TranslogWriter`]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/index/translog/TranslogWriter.java

Each translog is a sequence of files, each identified by a translog generation ID, each containing a sequence of operations, with the last file open for writes.
The last file has a part which has been fsync'ed to disk, and a part which has been written but not necessarily fsync'ed yet to disk.
Each operation is identified by a sequence number (`seqno`), which is monotonically increased by the engine's ingestion functionality.
Typically the entries in the translog are in increasing order of their sequence number, but not necessarily.
A [`Checkpoint`] file is also maintained, which is written on each fsync operation of the translog, and is necessary because it records important metadata and statistics about the translog, such as the current translog generation ID, its last fsync'ed operation and location (i.e., we should read only up to this location during recovery), the minimum translog generation ID, and the minimum and maximum sequence number of operations the sequence of translog generations include, all of which are used to identify the translog operations needed to be replayed upon recovery.
When the translog rolls over, e.g., upon the translog file exceeding a configurable size, a new file in the sequence is created for writes, and the last one becomes read-only.
A new commit flushed to the disk will also induce a translog rollover, since the operations in the translog so far will become eligible for truncation.

A few more words on terminology and classes used around the translog Java package.
A [`Location`] of an operation is defined by the translog generation file it is contained in, the offset of the operation in that file, and the number of bytes that encode that operation.
An [`Operation`] can be a document indexed, a document deletion, or a no-op operation.
A [`Snapshot`] iterator can be created to iterate over a range of requested operation sequence numbers read from the translog files.
The [`sync()`] method is the one that fsync's the current translog generation file to disk, and updates the checkpoint file with the last fsync'ed operation and location.
The [`rollGeneration()`] method is the one that rolls the translog, creating a new translog generation, e.g., called during an index flush.
The [`createEmptyTranslog()`] method creates a new translog, e.g., for a new empty index shard.
Each translog file starts with a [`TranslogHeader`] that is followed by translog operations.

Some internal classes used for reading and writing from the translog are the following.
A [`TranslogReader`] can be used to read operation bytes from a translog file.
A [`TranslogSnapshot`] can be used to iterate operations from a translog reader.
A [`MultiSnapshot`] can be used to iterate operations over multiple [`TranslogSnapshot`]s.
A [`TranslogWriter`] can be used to write operations to the translog.

#### Real-time GETs from the translog

[Get API]:https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-get.html
[`LiveVersionMap`]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/index/engine/LiveVersionMap.java

The [Get API] (and by extension, the multi-get API) supports a real-time mode, which can query documents by ID, even recently ingested documents that have not yet been refreshed and not searchable.
This capability is facilitated by another data structure, the [`LiveVersionMap`], which maps recently ingested documents by their ID to the translog location that encodes their indexing operation.
That way, we can return the document by reading the translog operation.

The tracking in the version map is not enabled by default.
The first real-time GET induces a refresh of the index shard, and a search to get the document, but also enables the tracking in the version map for newly ingested documents.
Thus, next real-time GETs are serviced by going first through the version map, to query the translog, and if not found there, then search (refreshed data) without requiring to refresh the index shard.

On a refresh, the code safely swaps the old map with a new empty map.
That is because after a refresh, any documents in the old map are now searchable in Lucene, and thus we do not need them in the version map anymore.

### Index Version

[IndexVersion]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/IndexVersion.java

[IndexVersions]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/IndexVersions.java

The [IndexVersion] tracks the on-disk format of an index. It is conceptually similar to [TransportVersion]
(which controls wire-protocol compatibility between nodes) but instead targets how index data and metadata are
serialized to Lucene files. Every time the serialization format of mappings, postings, doc values, or any other
persisted index structure changes, a new `IndexVersion` constant must be added.

The `IndexVersion` class contains two fields: an integer `id` and the
Lucene [Version](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/util/Version.html)
that the index was written with. The stored Lucene version is used for Lucene API calls that depend on the version,
such as [reading segment metadata](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/common/lucene/Lucene.java#L173).

The `IndexVersion` class was [introduced](https://github.com/elastic/elasticsearch/pull/94827) in 8.8.0. Before that,
the node release `Version` was used for both purposes. Prior to 8.9.0 the `id` field was the same as the release version,
for backwards compatibility. In 8.9.0 it changed to an incrementing number, and disconnected from the release version.

All known versions are declared as constants in [IndexVersions] (e.g. `UPGRADE_TO_LUCENE_10_4_0`,
`SEMANTIC_TEXT_FIELD_TYPE`, or `GENERIC_DENSE_VECTOR_FORMAT`).
This [list](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/IndexVersions.java#L62)
serves as the source of truth for version-to-Lucene mappings and merge-conflict checks.

The `IndexVersion`
is [stamped](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateIndexService.java#L1231)
on every index at creation time via
the [IndexMetadata](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/metadata/IndexMetadata.java)
`index.version.created` setting. The setting is [`Property.PrivateIndex`](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/metadata/IndexMetadata.java#L374): the cluster assigns it at creation and clients cannot set it explicitly. This version is immutable for the lifetime of the index and determines which code
paths are used when reading its data
([PostRecoveryMerger optimization example](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/indices/PostRecoveryMerger.java#L100)).

A separate `index.version.compatibility`
[setting](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/metadata/IndexMetadata.java#L394)
(defaulting to `index.version.created`)
can [be set](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateIndexService.java#L1874)
to a newer version to opt in to newer behavior while retaining backward-compatible defaults. For example, when
restoring a snapshot of an index with an `index.version.created` older than `MINIMUM_READONLY_COMPATIBLE`,
[RestoreService.convertLegacyIndex()](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/snapshots/RestoreService.java#L1801)
sets `index.version.compatibility` to the minimum index version supported by the cluster's nodes and adds
a write block, so that the index can be opened as a read-only archive. The subsequent paragraphs contain more details
about index version compatibility.

Elasticsearch supports a two-major-version compatibility window for index data. Indices created in the current
major version (N) or the previous major version (N-1) are fully supported for reading and writing. The minimum
index version for this full support is defined by
[IndexVersions.MINIMUM_COMPATIBLE](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/IndexVersions.java#L268).
Note that Lucene [enforces similar constraints](https://github.com/apache/lucene/issues/10708).

Indices created in the penultimate major version (N-2), i.e. older than `MINIMUM_COMPATIBLE` but at or above
[IndexVersions.MINIMUM_READONLY_COMPATIBLE](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/IndexVersions.java#L269),
can only be opened in read-only mode. They must have been marked as read-only (with a write block) on the previous
major version before upgrading. Their mappings and field types are still fully understood by the current version. Only
writes are blocked.

Indices older than `MINIMUM_READONLY_COMPATIBLE` are classified as legacy. Their on-disk format
is too old for the current version to fully understand their mappings or field types, so they are automatically converted
to degraded read-only archives via [RestoreService.convertLegacyIndex()](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/snapshots/RestoreService.java#L1801)
when restored from a snapshot.
Both `MINIMUM_COMPATIBLE` and `MINIMUM_READONLY_COMPATIBLE` are bumped with each new major release to maintain this
window.

#### Enforcement of IndexVersion compatibility in a cluster

Index version acts as a cluster-wide contract. New indices must use a format every data and master-eligible node
can read and write. A node is only admitted to the cluster if it can open every existing index. Allocation must not
move shards to nodes running an older Lucene version that cannot read segments already written by a newer one.

The cluster state's `DiscoveryNodes` field (see
the [Cluster State section](#cluster-state) for more details) computes and
records `maxDataNodeCompatibleIndexVersion`, which
is [the minimum of](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/node/DiscoveryNodes.java#L855)
all data and master-eligible node's `versionInfo.maxIndexVersion()`. This is the
highest index version the whole cluster supports.

[Creating an index](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateIndexService.java#L1128)
sets `index.version.created` to the minimum of `IndexVersion.current()` and
`getMaxDataNodeCompatibleIndexVersion()`. During a rolling upgrade, old and new nodes coexist: the cluster-wide value
stays at the lower ceiling until the last pre-upgrade node is gone, so indices created mid-upgrade get the same index
version as before the upgrade. Code paths and features that are gated on index version therefore stay on
backward-compatible behavior until the cluster is fully upgraded.

[A joining node](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/coordination/NodeJoinExecutor.java#L405)
is validated by `NodeJoinExecutor.ensureIndexCompatibility`: every index in metadata (open or closed) must fall within
that node's supported index-version range. Otherwise the join fails: for example, a node whose maximum index version is
too low cannot join while indices exist that were created with a newer format, and indices older than the joiner's
minimum writable version are only allowed if they still qualify as read-only-supported on that node.

The stamped `index.version.created` value ensures the index's on-disk format and feature gates are within the range
supported by the whole cluster. However, it does not pin the Lucene segment format: during a rolling upgrade, a shard
hosted on a newer node may flush or merge segments in a Lucene format that older nodes cannot read, even though they
still support the index's `IndexVersion`. To prevent allocating such shards onto incompatible nodes, [IndexVersionAllocationDecider](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/routing/allocation/decider/IndexVersionAllocationDecider.java)
only allows shard movement/recovery from a source node to a target node whose `DiscoveryNode#getMaxIndexVersion` is
equal-or-newer than the source node's. In other words, primary relocation and replica allocation are permitted only if
`target.maxIndexVersion >= source.maxIndexVersion`.

For more details on how index format compatibility interacts with upgrades,
see the [Index Format Backwards Compatibility](https://github.com/elastic/elasticsearch/blob/main/docs/internal/GeneralArchitectureGuide.md#index-format-backwards-compatibility)
section in the General Architecture Guide.

### Lucene

[Apache Lucene](https://lucene.apache.org/) is the search and indexing library at the core of every Elasticsearch shard.
Each shard is a Lucene index. Instead of treating Lucene as an opaque storage engine at the bottom of the
stack, Elasticsearch deeply integrates with it. It customizes file I/O, codecs, merge behavior, and reader
lifecycle.

For more details on how Elasticsearch wraps Lucene's `Directory` for file I/O, see the [Store](#store) section.
For how the `IndexWriter` and `DirectoryReader` are managed, including the distinction between "refresh"
(NRT reopen) and "flush" (durable Lucene commit), see [Segments, Refresh, and Flush](#segments-refresh-and-flush).

#### Lucene File Layout

A Lucene index on disk is a collection of segment files plus a
`segments_N` [info file](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/index/SegmentInfos.html) that
records which segments belong to the latest commit. Each segment is a self-contained, immutable mini-index composed
of several file types, each responsible for a different data structure:

| Extension | Content |
|-----------|---------|
| `.si`     | Segment metadata (doc count, unique id, diagnostics, Lucene version). |
| `.fnm`    | Field infos: field names, types (string, numeric, etc.), and index options (stored, indexed, doc values). |
| `.fdm`, `.fdt` | Stored fields metadata and data (original JSON `_source`, stored field values). |
| `.tim`, `.tip`, `.doc`, `.pos`, `.pay` | The inverted index: term dictionary, postings lists, positions, and payloads. |
| `.dvd`, `.dvm` | Doc values data and metadata (columnar storage for sorting, aggregations, scripting). |
| `.tvd`, `.tvx` | Term vectors data and index. |
| `.nvm`, `.nvd` | Norms (per-field length normalization factors used in scoring). |
| `.liv`    | Live documents bitset (tracks which docs have been deleted within the segment). |
| `.vec`, `.vex`, `.vem` | KNN vector data and graph (HNSW or other ANN structure). |
| `.kdm`, `.kdi`, `.kdd` | Points / BKD tree (numeric range queries, geo). |

Check out the
Lucene [documentation](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/codecs/lucene104/package-summary.html)
for more details on each of those specific files. Elasticsearch also maintains a registry of all known Lucene file
extensions in
[LuceneFilesExtensions](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/store/LuceneFilesExtensions.java).

The `segments_N` file and the `write.lock` file live at the directory root. A commit atomically publishes a new
`segments_N+1` as the active commit point, making the new set of segments visible.
The old segment files are removed once they are no longer
[referenced](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/index/IndexReader.html#decRef()) by any open
`IndexReader` or any retained commits. Elasticsearch's
[CombinedDeletionPolicy](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/CombinedDeletionPolicy.java), which implements Lucene's
[IndexDeletionPolicy](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/index/IndexDeletionPolicy.html),
manages which commits are retained. All commits more recent than
the [safe commit](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/CombinedDeletionPolicy.java#L55)
(the most recent commit whose max sequence number is at most the global checkpoint, used as the starting point
for peer recovery) are preserved. Older commits can also
be [pinned](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/CombinedDeletionPolicy.java#L216)
by external consumers (e.g., snapshot operations). `CombinedDeletionPolicy` also
[communicates the safe commit checkpoint information](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/CombinedDeletionPolicy.java#L109)
with the translog deletion policy.

#### Codecs

[PerFieldMapperCodec]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/codec/PerFieldMapperCodec.java
[Mapper]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/mapper/Mapper.java
[PerFieldFormatSupplier]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/codec/PerFieldFormatSupplier.java
[CodecService]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/codec/CodecService.java
[TSDBSyntheticIdPostingsFormat]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/codec/tsdb/TSDBSyntheticIdPostingsFormat.java
[TSDBSyntheticIdStoredFieldsReader]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/codec/tsdb/TSDBSyntheticIdStoredFieldsReader.java

Lucene's [Codec](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/codecs/package-summary.html) abstraction
lets each field type choose its own on-disk format. A codec is a bundle of format implementations: one for postings,
one for stored fields, one for doc values, one for vectors, etc.

Elasticsearch ships its own codec stack, managed by the [CodecService] class. It extends the current Lucene codec
(e.g. [Lucene104Codec](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/codecs/lucene104/Lucene104Codec.html))
with Elasticsearch-specific customizations:

- The [PerFieldMapperCodec] delegates postings, doc values, and KNN vector format selection
  to each field's [Mapper], allowing field types to choose specialized formats (See the [PerFieldFormatSupplier] for
  specific examples).
- The [DeduplicateFieldInfosCodec](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/codec/CodecService.java#L115)
  [wraps](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/codec/CodecService.java#L93)
  codecs registered in `CodecService` to
  deduplicate [FieldInfos](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/index/FieldInfo.html)
  metadata objects loaded from each segment's `.fnm` file. Field names, attribute keys, and attribute maps that are
  identical across segments are [interned](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/codec/DeduplicatingFieldInfosFormat.java#L50)
  so they share a single object in memory, reducing heap usage on indices with many segments.
- For time-series indices with synthetic IDs enabled (`index.mapping.synthetic_id`), the `AbstractTSDBSyntheticIdCodec`
  [avoids writing](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/mapper/SyntheticIdField.java#L32)
  the `_id` inverted index and stored field to disk. The `_id` field is still declared as indexed, but
  [SyntheticIdField](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/mapper/SyntheticIdField.java)
  produces an empty token stream so no postings are actually written on flush.
  At read time, the [TSDBSyntheticIdPostingsFormat] reconstructs postings from `_tsid`, `@timestamp`, and
  `_ts_routing_hash` doc values, and the [TSDBSyntheticIdStoredFieldsReader] synthesizes `_id` stored-field values on
  the fly from the same doc values. An
  [ES94BloomFilterDocValuesFormat](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/codec/bloomfilter/ES94BloomFilterDocValuesFormat.java)
  bloom filter is also built for the `_id` field so that existence checks during indexing can avoid `_tsid` and
  `@timestamp` doc-values lookups per document. The synthetic ID feature reduces the storage overhead for
  high-cardinality time-series indices.

As the [IndexVersion](#index-version) advances with each Lucene upgrade, new codec implementations are
introduced (e.g. [Elasticsearch92Lucene103Codec](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/codec/Elasticsearch92Lucene103Codec.java)). Older segments written by previous codecs remain readable
because Lucene's [SPI](https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/util/ServiceLoader.html)
mechanism loads the codec that originally wrote each segment.

#### Segment Merges

[MergePolicyConfig]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/MergePolicyConfig.java
[ThreadPoolMergeScheduler]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/ThreadPoolMergeScheduler.java
[SoftDeletesPolicy]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/SoftDeletesPolicy.java
[PrunePostingsMergePolicy]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/PrunePostingsMergePolicy.java
[PruningMergePolicy]:https://github.com/elastic/elasticsearch/blob/6b105b2e26988377532837e0360ddaaef87d279f/server/src/main/java/org/elasticsearch/index/engine/PruningMergePolicy.java
[ShuffleForcedMergePolicy]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/ShuffleForcedMergePolicy.java
[ThreadPoolMergeExecutorService]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/ThreadPoolMergeExecutorService.java

Because Lucene segments are immutable, updates and deletes cannot modify documents in place. Over time,
small segments accumulate from successive flushes.
[Merges](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/index/MergePolicy.html) combine multiple
segments into fewer, larger ones. Merges both reclaim space from obsolete documents and improve query performance by
reducing the number of segments a search must visit.

[MergePolicy](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/index/MergePolicy.html)s decide
which segments to merge and when. Elasticsearch configures this through [MergePolicyConfig], which uses
Lucene's [TieredMergePolicy](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/index/TieredMergePolicy.html)
as the default base policy, and
[LogByteSizeMergePolicy](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/index/LogByteSizeMergePolicy.html)
for time-series indices. Then the `InternalEngine`
[wraps](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java#L2808)
this base policy with several layers:

- [SoftDeletesRetentionMergePolicy](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/index/SoftDeletesRetentionMergePolicy.html):
  retains soft-deleted documents (more details about soft deletes in subsequent paragraphs).
- [PrunePostingsMergePolicy]: drops the `_id` field postings for deleted documents during segment merges, maintaining
  consistent update performance even when a large number of soft deleted/updated documents are retained.
- [PruningMergePolicy]: prunes `_recovery_source` stored fields and `_seq_no`-related fields for documents
  that no longer need them for replication.
- [ShuffleForcedMergePolicy]: randomizes segment ordering during force merges so that recently-indexed
  documents are not always co-located at the end, improving the efficiency of time-based queries.

[MergeScheduler](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/index/MergeScheduler.html)s decide how
merges are executed. The default implementation is [ThreadPoolMergeScheduler], which runs merges on the
[ThreadPoolMergeExecutorService] shared node-level thread pool with disk-aware I/O throttling.

Elasticsearch never hard-deletes documents from Lucene. Instead, both updates and deletes
use Lucene's [soft-delete](https://issues.apache.org/jira/browse/LUCENE-8198) mechanism. The [InternalEngine]
[calls](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java#L1645)
[IndexWriter.softUpdateDocument](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/index/IndexWriter.html#softUpdateDocument(org.apache.lucene.index.Term,java.lang.Iterable,org.apache.lucene.document.Field...)),
which atomically marks the previous version of a document as soft-deleted
(by setting a `__soft_deletes` numeric doc-values field to `1`) and writes the new version or a delete
tombstone into the current segment.
The old document is also marked as deleted in the `.liv` bitset, just like a hard delete. The soft-delete update
triggers a new generational `.liv` file for the segment that contained the previous version of the document.
The difference between hard and soft deletes is that hard-deleted documents are always discarded during merges,
whereas documents carrying the `__soft_deletes` marker can be selectively retained. Lucene's
[SoftDeletesRetentionMergePolicy](https://lucene.apache.org/core/10_4_0/core/org/apache/lucene/index/SoftDeletesRetentionMergePolicy.html)
decides which soft-deleted documents to keep by evaluating a retention query at merge time.
Elasticsearch provides this query through its [SoftDeletesPolicy], which evaluates whether a document can be discarded
based on the minimum sequence number to retain (evaluated from the global checkpoint), the
`index.soft_deletes.retention.operations` setting, and any active retention leases held by replicas or CCR followers.

Soft deletes are an essential feature for [CCR](#cross-cluster-replication-ccr).
CCR followers replay operations from their leader Lucene index to stay up to date. Without soft deletes, the leader
Lucene index could discard deleted documents during merges, leaving no trace that the delete operation ever occurred.
CCR followers that fall behind would no longer be able to catch up via operation replay. Soft-deleted tombstones
prevent this by preserving delete operations in the Lucene index until all followers have processed them.

# Recovery

When a shard is created on a node, it starts out empty. *Recovery* is the process of loading the shard's data from
some data source into the newly created `IndexShard` in order to make it available for index or search requests.
When the shard allocation process has chosen a node for a shard, it records its choice by writing an updated [ShardRouting][]
record into the cluster state's [IndexRoutingTable][]. The [ShardRouting][] entry includes [RecoverySource][] metadata that
describes where the shard's data can be found based on the shard's previous allocation. For example, a shard for a newly created
index will have its `recoverySource` set to `EMPTY_STORE` to indicate that recovery should bring up the shard without loading
any existing data, while a `recoverySource` of `EXISTING_STORE` would tell recovery to load the shard from files already
present on disk, likely because the node was restarted and had hosted the shard until it shut down.

The `IndicesClusterStateService` on each node listens for updates to the `IndexRoutingTable` and when it finds that a
shard in state `INITIALIZING` has been assigned to its node, it creates a fresh `IndexShard` for the assigned shard and kicks off a recovery process
for that node, using the [RecoverySource][] in the [ShardRouting][] entry to determine the parameters of the recovery process.
The full list of recovery types is defined in [RecoverySource.Type][]. The various modes are discussed below, roughly in
order of complexity. Some modes build on others; for example, snapshot recovery sets up a local data store by copying
files from a snapshot source and then uses `EXISTING_STORE` recovery. Similarly, if there is any local data, then
peer recovery starts by using  `EXISTING_STORE` recovery to bring the local shard as close to up to date as it can, and then
finishes synchronizing the shard through RPCs (Remote Procedure Calls) to an active source shard.

At the end of the recovery process, recovery finalization marks the shard as `STARTED` in cluster state, which makes it
available to handle index and search requests.

### Create a New Shard

The simplest form of recovery is `EMPTY_STORE`, which is just what it sounds like. This recovery type causes the recovery
process to invoke [StoreRecovery] for the recovery. `StoreRecovery` will create an empty shard directory on disk (deleting
any existing files that may be present) and bootstrap an empty translog and then tell the Engine to use that directory.
That's pretty much the whole process.

### Restore from an Existing Directory

Only slightly more complex is `EXISTING_STORE` recovery, which is used when the node is expected to have an up-to-date
copy of the shard's data already present on disk. Once again `StoreRecovery` manages the recovery process, but in this
case it expects to find Lucene files and a translog already present. It validates the Lucene files, potentially dropping
any incomplete commits that may not have been fsynced to disk before shutdown, tells the engine to open the directory
as its backing store, and then replays the transaction log to replay any operations that may have been acknowledged to
the client but not written into a durable Lucene commit. At that point the shard is ready to serve requests.

### Snapshot Recovery

In snapshot recovery, the data source is a snapshot of the index shard stored on a remote repository. Snapshot recovery,
also managed in `StoreRecovery` but invoked through the `recoverFromRepository` method, downloads and unpacks the snapshot
from the remote repository into the local shard directory and then invokes the same logic as `EXISTING_STORE` to bring
the shard on line, with some small differences. The key one is that the snapshot is taken from a Lucene commit, and
so it does not need to store the shard translog when the snapshot is taken, or restore it during recovery. Instead
it creates a new empty translog before bringing the shard on line.

### Local Shards Recovery
Local shards recovery is a type of recovery that reuses existing data from other shard(s) allocated on the current node (hence local shards). It is used exclusively to implement index [Shrink/Split/Clone APIs](#shrinksplitclone-index-apis).

This recovery type uses `HardlinkCopyDirectoryWrapper` to hard link or copy data from the source shard(s) directory. Copy is used if the runtime environment does not support hard links (e.g., on Windows). Source shard(s) directories are added using the `IndexWriter#addIndexes` API. Once an `IndexWriter` is correctly set up with source shard(s), the necessary data modifications are performed (like deleting excess documents during split) and a new commit is created for the recovering shard. After that recovery proceeds using standard store recovery logic utilizing the commit that was just created.

### Peer Recovery

Whereas the other recovery modes are used to bring up a *primary* shard, peer recovery is used to add replicas of an
already active primary shard. Peer recovery is also used for primary shard relocation, but relocation goes through
the standard peer recovery process to bring a replica in sync, before handing off the primary role to the recovered
replica during finalization.

Unlike `StoreRecovery`, peer recovery is managed through a separate service on the node recovering the shard, the
[PeerRecoveryTargetService][]. When the IndexShard sees that its recovery source is of type `PEER`, it hands over the
recovery process to `PeerRecoveryTargetService` by invoking its `startRecovery` method. This service begins by creating
an in-memory record of the recovery process to track its progress, and then runs `EXISTING_STORE` recovery in case the
recovering replica held a copy of the shard before that has gone out of sync (e.g., because the node holding the
replica restarted). Because the shard is a replica, it only recovers up to the latest known global checkpoint for the shard
and discards any operations in the local store that are ahead of that point (see [Translog][#Translog] for details).
Once the local shard has been brought close to current, the service then sends a request to a corresponding service on the source node, `PeerRecoverySourceService`,
to complete synchronization.

Synchronization begins by discovering any differences between the source and target shards and transmitting any missing files
to the target shard. The source for the files can be the source shard itself, but if a snapshot of the shard is available
that has some subset of the files to be transmitted, then recovery will fetch them from the snapshot in order to reduce
load on the source shard.

The next step is to transfer any operations from the source translog. Since the source shard is active,
it may be receiving index operations while recovery is in process. So, to ensure that the target shard doesn't miss any
new operations, the source shard adds the target to the shard's replication group (see the [replication][#Replication] docs)
*before* completing the operation transfer phase. Because of this ordering, any operations accepted on the shard between the
time it reads and sends the latest operation in the translog and the time the replica completes recovery are sent through the
request replication process and will not be lost. Once the target has been added to the recovery group, the source reads the
latest sequence number from its transaction log knowing that any updates past that will be handled by recovery, and replays
the translog to the target up to that point.

At this point the target is ready to be started as an in sync replica. However, peer recovery is also used to perform
primary relocation. If the target shard is being recovered in order to take over as primary, then the finalization
stage will call `IndexShard.relocate` to complete the handoff of primary responsibilities. This method blocks operations
on the source shard and sends an RPC to the target shard with the [ReplicationTracker.PrimaryContext][] needed
to promote the target to primary. Once the target acknowledges the handoff, the source shard moves itself into
replica mode.

[ShardRouting]:https://github.com/elastic/elasticsearch/blob/1d4a20ae194ce71fd5819786ba6dfb154ceb123f/server/src/main/java/org/elasticsearch/cluster/routing/ShardRouting.java
[IndexRoutingTable]:https://github.com/elastic/elasticsearch/blob/473c4da497681c889728c05cebb27030ae97fc13/server/src/main/java/org/elasticsearch/cluster/routing/IndexRoutingTable.java
[RecoverySource]:https://github.com/elastic/elasticsearch/blob/5346213ade708c63021824ad70cc3fa89f1ea307/server/src/main/java/org/elasticsearch/cluster/routing/RecoverySource.java
[RecoverySource.Type]:https://github.com/elastic/elasticsearch/blob/5346213ade708c63021824ad70cc3fa89f1ea307/server/src/main/java/org/elasticsearch/cluster/routing/RecoverySource.java#L79
[IndicesClusterStateService]:https://github.com/elastic/elasticsearch/blob/5346213ade708c63021824ad70cc3fa89f1ea307/server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java
[StoreRecovery]:https://github.com/elastic/elasticsearch/blob/d70878f488dfa2e2ba4d02e335c15be7cd4d5af2/server/src/main/java/org/elasticsearch/index/shard/StoreRecovery.java
[PeerRecoveryTargetService]:https://github.com/elastic/elasticsearch/blob/5346213ade708c63021824ad70cc3fa89f1ea307/server/src/main/java/org/elasticsearch/indices/recovery/PeerRecoveryTargetService.java
[ReplicationTracker.PrimaryContext]:https://github.com/elastic/elasticsearch/blob/1352df3f0b5157ca1d730428ea5aba2a7644e79b/server/src/main/java/org/elasticsearch/index/seqno/ReplicationTracker.java#L1573

# Data Tiers

(Frozen, warm, hot, etc.)

# Allocation

### Indexes and Shards

Each index consists of a fixed number of primary shards. The number of primary shards cannot be changed for the lifetime of the index. Each
primary shard can have zero-to-many replicas used for data redundancy. The number of replicas per shard can be changed dynamically.

The allocation assignment status of each shard copy is tracked by its [ShardRoutingState][]. The `RoutingTable` and `RoutingNodes` objects
are responsible for tracking the data nodes to which each shard in the cluster is allocated: see the [routing package javadoc][] for more
details about these structures.

[routing package javadoc]: https://github.com/elastic/elasticsearch/blob/v9.0.0-beta1/server/src/main/java/org/elasticsearch/cluster/routing/package-info.java
[ShardRoutingState]: https://github.com/elastic/elasticsearch/blob/4c9c82418ed98613edcd91e4d8f818eeec73ce92/server/src/main/java/org/elasticsearch/cluster/routing/ShardRoutingState.java#L12-L46

### Core Components

The `DesiredBalanceShardsAllocator` is what runs shard allocation decisions. It leverages the `DesiredBalanceComputer` to produce
`DesiredBalance` instances for the cluster based on the latest cluster changes (add/remove nodes, create/remove indices, load, etc.). Then
the `DesiredBalanceReconciler` is invoked to choose the next steps to take to move the cluster from the current shard allocation to the
latest computed `DesiredBalance` shard allocation. The `DesiredBalanceReconciler` will apply changes to a copy of the `RoutingNodes`, which
is then published in a cluster state update that will reach the data nodes to start the individual shard recovery/deletion/move work.

The `DesiredBalanceReconciler` is throttled by cluster settings, like the max number of concurrent shard moves and recoveries per cluster
and node: this is why the `DesiredBalanceReconciler` will make, and publish via cluster state updates, incremental changes to the cluster
shard allocation. The `DesiredBalanceShardsAllocator` is the endpoint for reroute requests, which may trigger immediate requests to the
`DesiredBalanceReconciler`, but asynchronous requests to the `DesiredBalanceComputer` via the `ContinuousComputation` component. Cluster
state changes that affect shard balancing (for example index deletion) all call some reroute method interface that reaches the
`DesiredBalanceShardsAllocator` to run reconciliation and queue a request for the `DesiredBalancerComputer`, leading to desired balance
computation and reconciliation actions. Asynchronous completion of a new `DesiredBalance` will also invoke a reconciliation action, as will
cluster state updates completing shard moves/recoveries (unthrottling the next shard move/recovery).

The `ContinuousComputation` saves the latest desired balance computation request, which holds the cluster information at the time of that
request, and a thread that runs the `DesiredBalanceComputer`. The `ContinuousComputation` thread takes the latest request, with the
associated cluster information, feeds it into the `DesiredBalanceComputer` and publishes a `DesiredBalance` back to the
`DesiredBalanceShardsAllocator` to use for reconciliation actions. Sometimes the `ContinuousComputation` thread's desired balance
computation will be signalled to exit early and publish the initial `DesiredBalance` improvements it has made, when newer rebalancing
requests (due to cluster state changes) have arrived, or in order to begin recovery of unassigned shards as quickly as possible.

### Rebalancing Process

There are different priorities in shard allocation, reflected in which moves the `DesiredBalancerReconciler` selects to do first given that
it can only move, recover, or remove a limited number of shards at once. The first priority is assigning unassigned shards, primaries being
more important than replicas. The second is to move shards that violate any rule (such as node resource limits) as defined by an
`AllocationDecider`. The `AllocationDeciders` holds a group of `AllocationDecider` implementations that place hard constraints on shard
allocation. There is a decider, `DiskThresholdDecider`, that manages disk memory usage thresholds, such that further shards may not be
allowed assignment to a node, or shards may be required to move off because they grew to exceed the disk space; or another,
`FilterAllocationDecider`, that excludes a configurable list of indices from certain nodes; or `MaxRetryAllocationDecider` that will not
attempt to recover a shard on a certain node after so many failed retries. The third priority is to rebalance shards to even out the
relative weight of shards on each node: the intention is to avoid, or ease, future hot-spotting on data nodes due to too many shards being
placed on the same data node. Node shard weight is based on a sum of factors: disk memory usage, projected shard write load, total number
of shards, and an incentive to distribute shards within the same index across different nodes. See the `WeightFunction` and
`NodeAllocationStatsAndWeightsCalculator` classes for more details on the weight calculations that support the `DesiredBalanceComputer`
decisions.

### Inter-Node Communication

The elected master node creates a shard allocation plan with the `DesiredBalanceShardsAllocator` and then selects incremental shard
movements towards the target allocation plan with the `DesiredBalanceReconciler`. The results of the `DesiredBalanceReconciler` is an
updated `RoutingTable`. The `RoutingTable` is part of the cluster state, so the master node updates the cluster state with the new
(incremental) desired shard allocation information. The updated cluster state is then published to the data nodes. Each data node will
observe any change in shard allocation related to itself and take action to achieve the new shard allocation by: initiating creation of a
new empty shard; starting recovery (copying) of an existing shard from another data node; or removing a shard. When the data node finishes
a shard change, a request is sent to the master node to update the shard as having finished recovery/removal in the cluster state. The
cluster state is used by allocation as a fancy work queue: the master node conveys new work to the data nodes, which pick up the work and
report back when done.

- See `DesiredBalanceShardsAllocator#submitReconcileTask` for the master node's cluster state update post-reconciliation.
- See `IndicesClusterStateService#doApplyClusterState` for the data node hook to observe shard changes in the cluster state.
- See `ShardStateAction#sendShardAction` for the data node request to the master node on completion of a shard state change.

# Autoscaling

The Autoscaling API in ES (Elasticsearch) uses cluster and node level statistics to provide a recommendation
for a cluster size to support the current cluster data and active workloads. ES Autoscaling is paired
with an ES Cloud service that periodically polls the ES elected master node for suggested cluster
changes. The cloud service will add more resources to the cluster based on Elasticsearch's recommendation.
Elasticsearch by itself cannot automatically scale.

Autoscaling recommendations are tailored for the user [based on user defined policies][], composed of data
roles (hot, frozen, etc.) and [deciders][]. There's a public [webinar on autoscaling][], as well as the
public [Autoscaling APIs] docs.

Autoscaling's current implementation is based primary on storage requirements, as well as memory capacity
for ML and frozen tier. It does not yet support scaling related to search load. Paired with ES Cloud,
autoscaling only scales upward, not downward, except for ML nodes that do get scaled up _and_ down.

[based on user defined policies]: https://www.elastic.co/guide/en/elasticsearch/reference/current/xpack-autoscaling.html
[deciders]: https://www.elastic.co/guide/en/elasticsearch/reference/current/autoscaling-deciders.html
[webinar on autoscaling]: https://www.elastic.co/webinars/autoscaling-from-zero-to-production-seamlessly
[Autoscaling APIs]: https://www.elastic.co/guide/en/elasticsearch/reference/current/autoscaling-apis.html

### Plugin REST and TransportAction entrypoints

Autoscaling is a [plugin][]. All the REST APIs can be found in [autoscaling/rest/][].
`GetAutoscalingCapacityAction` is the capacity calculation operation REST endpoint, as opposed to the
other rest commands that get/set/delete the policies guiding the capacity calculation. The Transport
Actions can be found in [autoscaling/action/], where [TransportGetAutoscalingCapacityAction][] is the
entrypoint on the master node for calculating the optimal cluster resources based on the autoscaling
policies.

[plugin]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/Autoscaling.java#L72
[autoscaling/rest/]: https://github.com/elastic/elasticsearch/tree/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/rest
[autoscaling/action/]: https://github.com/elastic/elasticsearch/tree/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/action
[TransportGetAutoscalingCapacityAction]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/action/TransportGetAutoscalingCapacityAction.java#L82-L98

### How cluster capacity is determined

[AutoscalingMetadata][] implements [Metadata.ClusterCustom][] in order to persist autoscaling policies. Each
Decider is an implementation of [AutoscalingDeciderService][]. The [AutoscalingCalculateCapacityService][]
is responsible for running the calculation.

[TransportGetAutoscalingCapacityAction.computeCapacity] is the entry point to [AutoscalingCalculateCapacityService.calculate],
which creates a [AutoscalingDeciderResults][] for [each autoscaling policy][]. [AutoscalingDeciderResults.toXContent][] then
determines the [maximum required capacity][] to return to the caller. [AutoscalingCapacity][] is the base unit of a cluster
resources recommendation.

The `TransportGetAutoscalingCapacityAction` response is cached to prevent concurrent callers
overloading the system: the operation is expensive. `TransportGetAutoscalingCapacityAction` contains
a [CapacityResponseCache][]. `TransportGetAutoscalingCapacityAction.masterOperation`
calls [through the CapacityResponseCache][], into the `AutoscalingCalculateCapacityService`, to handle
concurrent callers.

[AutoscalingMetadata]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/AutoscalingMetadata.java#L38
[Metadata.ClusterCustom]: https://github.com/elastic/elasticsearch/blob/f461731a30a6fe55d7d7b343d38426ddca1ac873/server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java#L147
[AutoscalingDeciderService]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/capacity/AutoscalingDeciderService.java#L16-L19
[AutoscalingCalculateCapacityService]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/capacity/AutoscalingCalculateCapacityService.java#L43

[TransportGetAutoscalingCapacityAction.computeCapacity]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/action/TransportGetAutoscalingCapacityAction.java#L102-L108
[AutoscalingCalculateCapacityService.calculate]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/capacity/AutoscalingCalculateCapacityService.java#L108-L139
[AutoscalingDeciderResults]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/capacity/AutoscalingDeciderResults.java#L34-L38
[each autoscaling policy]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/capacity/AutoscalingCalculateCapacityService.java#L124-L131
[AutoscalingDeciderResults.toXContent]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/capacity/AutoscalingDeciderResults.java#L78
[maximum required capacity]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/capacity/AutoscalingDeciderResults.java#L105-L116
[AutoscalingCapacity]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/capacity/AutoscalingCapacity.java#L27-L35

[CapacityResponseCache]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/action/TransportGetAutoscalingCapacityAction.java#L44-L47
[through the CapacityResponseCache]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/action/TransportGetAutoscalingCapacityAction.java#L97

### Where the data comes from

The Deciders each pull data from different sources as needed to inform their decisions. The
[DiskThresholdMonitor][] is one such data source. The Monitor runs on the master node and maintains
lists of nodes that exceed various disk size thresholds. [DiskThresholdSettings][] contains the
threshold settings with which the `DiskThresholdMonitor` runs.

[DiskThresholdMonitor]: https://github.com/elastic/elasticsearch/blob/v8.13.2/server/src/main/java/org/elasticsearch/cluster/routing/allocation/DiskThresholdMonitor.java#L53-L58
[DiskThresholdSettings]: https://github.com/elastic/elasticsearch/blob/v8.13.2/server/src/main/java/org/elasticsearch/cluster/routing/allocation/DiskThresholdSettings.java#L24-L27

### Deciders

The `ReactiveStorageDeciderService` tracks information that demonstrates storage limitations are causing
problems in the cluster. It uses [an algorithm defined here][]. Some examples are
- information from the `DiskThresholdMonitor` to find out whether nodes are exceeding their storage capacity
- number of unassigned shards that failed allocation because of insufficient storage
- the max shard size and minimum node size, and whether these can be satisfied with the existing infrastructure

[an algorithm defined here]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/storage/ReactiveStorageDeciderService.java#L158-L176

The `ProactiveStorageDeciderService` maintains a forecast window that [defaults to 30 minutes][]. It only
runs on data streams (ILM, rollover, etc.), not regular indexes. It looks at past [index changes][] that
took place within the forecast window to [predict][] resources that will be needed shortly.

[defaults to 30 minutes]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/storage/ProactiveStorageDeciderService.java#L32
[index changes]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/storage/ProactiveStorageDeciderService.java#L79-L83
[predict]: https://github.com/elastic/elasticsearch/blob/v8.13.2/x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/storage/ProactiveStorageDeciderService.java#L85-L95

There are several more Decider Services, implementing the `AutoscalingDeciderService` interface.

# Snapshot / Restore

Snapshot copies index data files from node local storage to a remote repository. These files can later be restored
from the repository back to local storage to re-create the index. In addition to indices, it can also backup and
restore the cluster state [Metadata][] so that settings, templates, pipelines and other configurations can be preserved.

Snapshots are deduplicated in that it does not copy a data file if it has already been copied in a previous snapshot.
Instead, it adds a reference to the existing file in the metadata stored in the repository, effectively a ref-tracking
system for the data files. This also means we can freely delete any snapshot without worrying about affecting other
snapshots.

This [snapshots Java package documentation][] provides a good explanation on how snapshot operations work.

Restoring a snapshot is a process which largely relies on [index recoveries](#Recovery). The restore service initializes
the process by preparing shards of restore indices as unassigned with snapshot as their recovery source ([SnapshotRecoverySource][])
in the cluster state. These shards go through the regular allocation process to be allocated. They then recover on the target nodes
by copying data files from the snapshot repository to local storage.

Both snapshot and restore are coordinated by the master node, while index data transfer is done by the data nodes.
The communications from the master node to data nodes are always cluster state updates. Data nodes send transport
requests to the master node to update the status. These requests, at the end, also triggers cluster state updates
which can be further reacted upon by the data nodes until the processes are complete.

[Metadata]: https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java
[snapshots Java package documentation]: https://github.com/elastic/elasticsearch/blob/8ccbdd553b096465a282297332d2389328ed665a/server/src/main/java/org/elasticsearch/snapshots/package-info.java#L11
[SnapshotRecoverySource]: https://github.com/elastic/elasticsearch/blob/31db11b3b067baf97e305bfefefea9e11cb85371/server/src/main/java/org/elasticsearch/cluster/routing/RecoverySource.java#L207

### Snapshot Repository

A [Repository][] must be created before any snapshot operation can take place. There are different types of
repositories. The most common ones are file system based (FS) and cloud storage based (S3, GCS, Azure) which
all extend the [BlobStoreRepository][] class. A repository must be registered as writeable with a single
cluster while registered as readable with multiple clusters. NOTE registering a repository as writeable
with multiple clusters can lead to data corruption. We try our best to detect such situation, but it's not
completely foolproof.

The content structure of a repository is similar to the local index storage structure, with `indices` folder
holding indices data separate by their UUID and shard ID. Here is a simple example of a repository structure:

```
my-repository/
├── index-2       <-- The root blob file of the repository, also called repository generation file
├── index.latest
├── indices
│   └── 8K9JNuqjTnygrjKY8qmsiA   <-- UUID of the snapshotted index. Not the same index UUID in the cluster
│       ├── 0                    <-- shard 0
│       │   ├── __I0e0reaMQzuXv8fY1GYD2w           <-- data file
│       │   ├── __XqE3EVhOREWBnCHASLALtw           <-- another data file
│       │   ├── index-pPXvvdFWSmajXZcfrIwggA       <-- shard level generation file
│       │   └── snap-3kXOuTRmTFm6VMcEqPkKNQ.dat    <-- shard level snapshot info
│       └── meta-tdTzfI8BkmhGlJ2SvOok.dat          <-- index metadata
├── meta-3kXOuTRmTFm6VMcEqPkKNQ.dat    <-- cluster metadata
└── snap-3kXOuTRmTFm6VMcEqPkKNQ.dat    <-- snapshot information
```
See also the [blobstore Java package documentation][] for more details on the repository structure.

The most important file in the repository is the `index-N` file, where `N` is a numeric generation number starting
from `9`. Its corresponding Java class is [RepositoryData][].
This file holds the global state of the repository, including all valid snapshots and their corresponding
indices, shards and index metadata. Every mutable operation on the repository, such as creating or deleting a snapshot,
results in a new `index-N` file being created with an incremented generation number. The `index.latest` file stores
the latest repository generation and is effectively a pointer to the latest `index-N` file.

The repository is not rescuable if the repository generation file is corrupted. This is the reason that we are
very careful when updating this file by leveraging cluster consensus to ensure only the latest master node update
it to a generation that is accepted by the rest of the cluster members. The updating process also attempts to
detect concurrent writes to avoid multiple clusters writing to the same repository. This is done by comparing
the expected repository generation to the actual generation files in the repository. If this file is corrupted,
as reported occasionally on SDHs, it is almost certain that some other external process has modified it.

If other files in the repository are corrupted, we can usually delete the broken snapshots and retain the good ones.
The broken snapshots usually lead to exception being thrown when accessed by APIs like the [GetSnapshot API][].
Since v8.16.0, there is also [VerifyRepositoryIntegrity API][] that can be used to actively scan the repository
and identify any corrupted snapshots.

The state of a repository must always transition through fully valid states in that there are no dangling references
to non-existent blobs. This is an important property that guarantees the repository integrity as long as there is
no external interference. We add new blobs before they become reachable from the root, then update the root blob,
and only if the root-blob update succeeds do we delete any now-unreachable blobs. See also
[Creation of a Snapshot](#creation-of-a-snapshot) and [Deletion of a Snapshot](#deletion-of-a-snapshot) for more details.

It is worth note that repository's compatibility guarantee is more permissive than Elasticsearch's general version
compatibility policy. A repository may contain snapshots from a version as old as 5.0.0 and, if it does, the repository
layout must remain compatible (for reads and writes) with the oldest version in the repository so that these snapshots
remain restorable in the corresponding versions. With older snapshots deleted, the repository will start using
new format when possible (see also the static `IndexVersion` constants in `SnapshotsService`).

[Repository]: https://github.com/elastic/elasticsearch/blob/2d4687af9bf21321573eb64eade0b0365213a303/server/src/main/java/org/elasticsearch/repositories/Repository.java#L53
[BlobStoreRepository]: https://github.com/elastic/elasticsearch/blob/2d4687af9bf21321573eb64eade0b0365213a303/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java#L200
[blobstore Java package documentation]: https://github.com/elastic/elasticsearch/blob/24fad8fac774983bb231da34321108abef102745/server/src/main/java/org/elasticsearch/repositories/blobstore/package-info.java#L11
[RepositoryData]: https://github.com/elastic/elasticsearch/blob/31db11b3b067baf97e305bfefefea9e11cb85371/server/src/main/java/org/elasticsearch/repositories/RepositoryData.java#L58
[GetSnapshot API]: https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-snapshot-get
[VerifyRepositoryIntegrity API]: https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-snapshot-repository-verify-integrity

#### Repository Management

A snapshot repository has two components: (1) a `RepositoryMetadata` containing the configuration stored
as a `Metadata.ProjectCustom` in the cluster state; (2) the actual repository object created on the master
node and each data node. A series of APIs are available to create, update, get and delete repositories.
Each of these APIs is backed by a `TransportMasterNodeAction` which publishes a cluster state for the
`RepositoryMetadata` change so that relevant nodes can update their local repository objects accordingly.
The API call will only return after the repository object has been updated on all relevant nodes.
For creating a repository, it also performs extra verification steps which (1) attempts to create a temporary
repository directly on the master node as well as write and read a test file before proceeding with the
cluster state update and (2) performs another write and read test after repository objects are created on all
relevant nodes. This verification steps are enabled by default and can be disabled per request.

The core service class is [RepositoriesService][] which all APIs eventually delegate to. It also implements
`ClusterStateApplier` which performs the actual repository object creation, update and deletion.

Besides the APIs, reserved repositories are managed via file based settings. These repositories are managed
by Elasticsearch service providers, such as the Elastic Cloud. File based settings is effectively a way
to publish cluster state based on file contents. Hence, they also go through the same code path as the APIs
under the hood. Reserved repositories cannot be modified via APIs.

While new repository can be created at any time, deleting a repository has some restrictions. In general,
a repository cannot be deleted if it is in use by either ongoing snapshots or restores or hosting mounted
searchable snapshots. Since v9.4.0, the default snapshot repository cannot be deleted either. A default
repository is meant to be the repository used by ILM and SLM when no repository is explicitly specified.
Updating a repository usually involves closing the existing repository first and creating a new one.
Therefore, they often subject to the same restrictions as deletion.

Cloud storage backed (S3, GCS, Azure) repositories requires network access to the storage services. Hence,
they have concept of clients which manages the network requests. At least the `default` client must be
configured for a cluster which is used by a repository if no client is explicitly specified. Multiple
clients can be added, via `elasticsecrh.yml` to the same cluster and used by different repositories to
spread snapshots to different locations if so desired.

We allow different implementations of the same cloud storage type to be used as long as they are compatible
in both APIs and performance characteristics. For example, many storage service claims S3 compatibility.
But they may fall short under load or even just return outright incorrect responses in some corner cases.
The [RepositoryAnalyze API][] can be used to proactively test the
compatibility which we suggest on SDHs from time to time.

[RepositoriesService]: https://github.com/elastic/elasticsearch/blob/f55a90c941f5ca80fff4978be7b15e97614ce55f/server/src/main/java/org/elasticsearch/repositories/RepositoriesService.java#L98
[PutRepository API]: https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-snapshot-create-repository
[RepositoryAnalyze API]: https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-snapshot-repository-analyze

### Creation of a Snapshot

Creating a snapshot is the most involved snapshot operation because it requires work to be done on
both master and data nodes. In contrast, cloning snapshot, deleting snapshot and cleaning repository
all happen entirely on the master node.

The following Java classes are mostly responsible for the snapshot creation process:
1. [SnapshotsService][] - runs on the master node and coordinates the overall snapshot process.
2. [SnapshotsServiceUtils][] - utility class separated from `SnapshotsService` to reduce file length.
3. [SnapshotShardsService][] - runs on data node and manages the actual snapshot of individual shards.
4. [BlobStoreRepository][] - used by both master and data nodes to read and write snapshot data and metadata.

As discussed earlier, a snapshot operation always starts with a cluster state triggered by a transport request.
Such is the case for snapshot creation. When `SnapshotsService` receives the request, it computes all indices
and their shards required for the snapshot and creates an object representing this snapshot and stores it in the
cluster state. The overarching object representing snapshot creation in the cluster state is[SnapshotsInProgress][]
which is essentially a map keyed by repository name with values being a list of ongoing snapshots(`SnapshotsInProgress.Entry`)
for that repository. It's a list because the order is important: Snapshots within each repository operate as a queue,
such that each shard's snapshots run in order. Different snapshots may complete in different orders if they target different shards.
Each ongoing snapshot further tracks its overall state (`SnapshotsInProgress#State`)
as well states (`SnapshotsInProgress#ShardState`) of each shard required for the snapshot.

When the snapshot creation entry (`SnapshotsInProgress.Entry`) is first added to the cluster state, its shard states
can be in different `ShardState` depending on the shard's status and snapshot activities:
* If the shard is started and has no other active snapshot activity, its state is set to `INIT` indicating
  it is ready to be snapshotted by the data node hosting it. Shards in this state cannot relocate due to
  `SnapshotInProgressAllocationDecider`.
* If the shard is started but is still running another snapshot, its state is set to `QUEUED` indicating it
  will be snapshotted later when the ongoing snapshots and any other snapshots queued before it are finished.
* If the shard is relocating or initializing, its state is set to `WAITING` which will be changed to a new state
  once the shard finishes relocation or initialization.
* If the index no longer exists or the shard is unassigned, the shard state is set to `MISSING`. This state
  is final. A snapshot creation fails on shard with this state if it is issued with `partial=false`. Note
  snapshot creation is always issued with `partial=true` in Elastic Cloud so that snapshot does not fail
  entirely for temporary shard unavailability.
* If the node hosting the shard is being shutdown, the shard state is set to `PAUSED_FOR_NODE_REMOVAL`.
  This state will be updated when the node finishes the shutdown process or the shard state changes.

When a data node (`SnapshotShardsService`) receives the updated cluster state with a new snapshot entry,
it takes the shards with the `INIT` state and hosted on itself to create a shard snapshot task for each
of them. The shard state is computed for all shards involved in the snapshot at once when the snapshot entry
is created. A large snapshot can easily have thousands of shards with `INIT` state indicating ready to be snapshotted.
To avoid overwhelming the data nodes, a dedicated snapshot thread pool as well as `ThrottledTaskRunner` are
used to keep concurrent running shard snapshots under control. Priority is given for snapshots which started
earlier. We also order by shard to limit the number of incomplete shard snapshots (see also [ShardSnapshotTaskRunner][]
and `ShardSnapshotTaskRunner#COMPARATOR`).

The lifecycle of each shard snapshot is also tracked in-memory on the data node with `IndexShardSnapshotStatus`.
The status is indicated by `IndexShardSnapshotStatus#Stage` which is updated at various points during the process.
When a shard snapshot task runs, it first acquires an index commit of the shard so that the files to be copied remain
available throughout the shard snapshot process without being deleted by ongoing indexing activities. It then
writes a new shard level generation file (`index-<UUID>.data`, Java class [BlobStoreIndexShardSnapshots][]).
This is basically a shard level catalog file pointing to all the valid shard snapshots. Each snapshot creates a new one.
The UUID is used to avoid name collision.
Previous shard generation files are not deleted because they may still be needed if the current shard snapshot fails.
Following that, shard level data files are copied to the repository with `BlobStoreRepository#doSnapshotShard`,
`BlobStoreRepository#snapshotFile` etc.
After all data files are uploaded, it writes a shard level snapshot file (`snap-<UUID>.dat`, Java class
[BlobStoreIndexShardSnapshot][]) indicating what data files
should be included in this shard snapshot. Note the data file's physical name is replaced with double underscore (`__`)
followed by an UUID to avoid name collision. The actual name is mapped and stored in the shard level snapshot file.

It is worth note that the shard level generation file (`index-<UUID>.data`) can be reconstructed from all shard
level snapshot files (`snap-<UUID>.dat`) so that it is technically redundant. However, listing and reading
all shard snapshot files can be rather inefficient since there could be hundreds or thousands of these files.
Hence, having the shard level generation file helps with performance for deletion operations which
needs to read only a single file to decide what can be deleted at the shard level.

Once the shard snapshot is completed successfully, the data node releases the previously acquired index commit and
sends a transport request(`UpdateIndexShardSnapshotStatusRequest`) with the new shard generation (`ShardGeneration`)
to the master node to update its shard status (`SnapshotsInProgress#ShardState`) in the cluster state. If there is any
`QUEUED` shard snapshot for the same shard, the master node (`SnapshotsService`) updates the next one's
status to `INIT` so that it can run. The master responds to the data node only after the cluster state is published.

When all shards in a snapshot are completed, the master node performs a finalization step
(`SnapshotsService#SnapshotFinalization` and `BlobStoreRepository#finalizeSnapshot`) which does the following:
1. Create a `SnapshotInfo` object representing the completed snapshot for serialization.
2. Collect and write the latest `Metadata` and `IndexMetadata` relevant to this snapshot.
3. Write the snapshot metadata file `snap-<UUID>.dat` (Java class [SnapshotInfo][]).
4. Create a new root blob (repository generation file, `index-N`) with incremented generation number
   and updated content including the new snapshot and publish a cluster state to accept this new
   generation as the current/safe (`BlobStoreRepository#writeIndexGen`).

Step 4 is the most critical one. The root blob is intentionally written at the very last so that
any prior failure only leaves some redundant files in the repository which will be cleaned up in due time.
A snapshot is not completed until the root blob is successfully updated. To ensure consistency,
updating the root blob is a 3-steps process leveraging cluster consensus:
1. Picks a new pending repository generation number which is greater than the current pending generation
   and publishes a cluster state update for it. Both repository generation number and pending generation
   numbers are part of [RepositoryMetadata][].
2. If previous step is successful, writes the new root blob with the pending generation number.
3. Publishes another cluster state to set the current/safe repository generation to the new pending generation.

When master fails over during snapshot creation, the above steps ensures that only the new master can
successfully update the root blob to avoid data corruption.

Multiple snapshots can run concurrently in the same repository. But the process is sequential at shard level,
i.e. only one shard snapshot for the same shard can be in the `INIT` state at any time.
Snapshot deletions and creations are mutually exclusive. See also [Deletion of a Snapshot](#Deletion-of-a-Snapshot).

[SnapshotsService]: https://github.com/elastic/elasticsearch/blob/5c3270085a72ec6b97d2cd34e2a18e664ebd28ba/server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java#L133
[SnapshotsServiceUtils]: https://github.com/elastic/elasticsearch/blob/5c3270085a72ec6b97d2cd34e2a18e664ebd28ba/server/src/main/java/org/elasticsearch/snapshots/SnapshotsServiceUtils.java#L83
[SnapshotShardsService]: https://github.com/elastic/elasticsearch/blob/5c3270085a72ec6b97d2cd34e2a18e664ebd28ba/server/src/main/java/org/elasticsearch/snapshots/SnapshotShardsService.java#L81
[BlobStoreRepository]: https://github.com/elastic/elasticsearch/blob/2d4687af9bf21321573eb64eade0b0365213a303/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java#L200
[SnapshotsInProgress]: https://github.com/elastic/elasticsearch/blob/5c3270085a72ec6b97d2cd34e2a18e664ebd28ba/server/src/main/java/org/elasticsearch/cluster/SnapshotsInProgress.java#L78
[ShardSnapshotTaskRunner]: https://github.com/elastic/elasticsearch/blob/01ace3927df065f1caf090653404f29688e8103a/server/src/main/java/org/elasticsearch/repositories/blobstore/ShardSnapshotTaskRunner.java#L36
[BlobStoreIndexShardSnapshots]: https://github.com/elastic/elasticsearch/blob/495c7c2f4ec5817001aa767f6d45a9f1c8c31082/server/src/main/java/org/elasticsearch/index/snapshots/blobstore/BlobStoreIndexShardSnapshots.java#L40
[BlobStoreIndexShardSnapshot]: https://github.com/elastic/elasticsearch/blob/495c7c2f4ec5817001aa767f6d45a9f1c8c31082/server/src/main/java/org/elasticsearch/index/snapshots/blobstore/BlobStoreIndexShardSnapshot.java#L38
[SnapshotInfo]: https://github.com/elastic/elasticsearch/blob/495c7c2f4ec5817001aa767f6d45a9f1c8c31082/server/src/main/java/org/elasticsearch/snapshots/SnapshotInfo.java#L50
[RepositoryMetadata]: https://github.com/elastic/elasticsearch/blob/495c7c2f4ec5817001aa767f6d45a9f1c8c31082/server/src/main/java/org/elasticsearch/cluster/metadata/RepositoryMetadata.java#L24

#### Shard snapshot pausing

When a node is shutting down, it must vacate all shards via relocation. Since shards being snapshotted
(shard snapshot status `INIT`) cannot relocate, we need a way to transit these shards out of the
`INIT` state to avoid stalling the shutdown process. This is where the shard snapshot pausing mechanism
comes into play.

When a node shutdown is initiated, `SnapshotsService` reacts to the new shutdown metadata by updating
`SnapshotsInProgress#nodesIdsForRemoval` which tracks the node IDs for the shutting down node. When
this change is published and observed by the shutting down data node (`SnapshotShardsService`), it pauses
its shard snapshots by first setting the shard snapshot status `PAUSING`, which is checked regularly
by the file uploading process (`BlobStoreRepository#snapshotFile`) and leads to a `PausedSnapshotException`
to be thrown to abort the shard snapshot. The data node then notifies the master node about the status
change with the same status update request (`UpdateIndexShardSnapshotStatusRequest`) in the happy path.
The master node updates the shard state to `PAUSED_FOR_NODE_REMOVAL` upon receiving the notification.
From this point on, the shard can relocate as normal. When the shard is started on the target node,
`SnapshotsService` will observe the shard state changes and transition the shard snapshot status back to
`INIT`.

If a node is already being shutdown when a new snapshot creation request arrives, the relevant
shard snapshot will be created with `PAUSED_FOR_NODE_REMOVAL` as its initial state. This assumes there
is no ongoing shard snapshot that is already `PAUSED_FOR_NODE_REMOVAL`, in which case the new shard
snapshot will start out as `QUEUED`.

When the node shutdown completes and its associated shutdown metadata is removed from the cluster state,
`SnapshotsService` will also remove the node ID from `SnapshotsInProgress#nodesIdsForRemoval`.

### Deletion of a Snapshot

Both completed snapshots and ongoing snapshots can be deleted. Unless the snapshot being deleted has not
started yet, e.g. all its shards are in `QUEUED` state, which means it can be deleted right away from the
cluster state without touch the repository content, deletion must run exclusively in a repository.

If the deletion requires any file removal in the repository, `SnapshotsService` creates/updates the
`SnapshotDeletionsInProgress` in the cluster state to track the new deletion. If the snapshot is
currently running, it also updates any incomplete shard snapshots to `ShardState.ABORTED` for the
data node (`SnapshotShardsService`) to react once the cluster state is published. The data node goes through
a similar process to the shard snapshot pausing but with a different exception (`AbortedSnapshotException`)
to interrupt the shard snapshot process and sends a request back to the master node to update the
corresponding status (`ShardState.FAILED`) tracked in the cluster state. Once all shard snapshots
stop, deletion will proceed to remove relevant files from the repository as well as create a new
root blob (`index-N`) with the same mechanism described in the snapshot creation section.
File deletions (`BlobStoreRepository#SnapshotsDeletion`) happen entirely on the master node.

### Clone of a Snapshot

TODO: Clone is not used in Elastic Cloud Serverless.

### Cleaning a Repository
Repository clean up is cluster wide exclusive and must run by itself. It does not actually clean up anything
more than a regular snapshot deletion. It was useful in the early days when we had some long-since-fixed
leaks that needed cleaning up in ECH. It is not used in Elastic Cloud Serverless.

### Restoring a Snapshot

[RestoreService][] is responsible for handling the initial restore request and prepare the unassigned shards
with snapshot as their recovery source in the cluster state. Once the shard is allocated on a data node, the
recovery process kicks in and eventually calls into `IndexShard#restoreFromSnapshot` which delegates to
`BlobStoreRepository#restoreShard` to copy data files from the repository to local storage.

[RestoreService]: https://github.com/elastic/elasticsearch/blob/1b7e99ee7a4ec92d20846be639fa4c3d15f20abc/server/src/main/java/org/elasticsearch/snapshots/RestoreService.java#L149

### Detecting Multiple Writers to a Single Repository

This is a best effort attempt to prevent repository corruption due to concurrent writes from multiple clusters.
When writing the root blob (`index-N`), we cross compare the cached and expected repository generation
(see `BlobStoreRepository#latestKnownRepoGen` and `BlobStoreRepository#latestKnownRepositoryData`)
to the generation physically found in the repository. The writing also fails if the blob already exists
since this indicates that some other cluster attempted to write to the repository. We do that in all repositories
that support such a check, which includes FS/GCS/Azure since forever and S3 since 9.2 (but not HDFS).
A `RepositoryException` is thrown on any mismatch or conflicts. The repository is marked as corrupted
by setting its generation number to `RepositoryData.CORRUPTED_REPO_GEN` to block further write operations.

# Task Management / Tracking

[TransportRequest]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/transport/TransportRequest.java
[TaskManager]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/tasks/TaskManager.java
[TaskManager#register]:https://github.com/elastic/elasticsearch/blob/6d161e3d63bedc28088246cff58ce8ffe269e112/server/src/main/java/org/elasticsearch/tasks/TaskManager.java#L125
[TaskManager#unregister]:https://github.com/elastic/elasticsearch/blob/d59df8af3e591a248a25b849612e448972068f10/server/src/main/java/org/elasticsearch/tasks/TaskManager.java#L317
[TaskId]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/tasks/TaskId.java
[Task]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/tasks/Task.java
[TaskAwareRequest]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/tasks/TaskAwareRequest.java
[TaskAwareRequest#createTask]:https://github.com/elastic/elasticsearch/blob/6d161e3d63bedc28088246cff58ce8ffe269e112/server/src/main/java/org/elasticsearch/tasks/TaskAwareRequest.java#L50
[CancellableTask]:https://github.com/elastic/elasticsearch/blob/d59df8af3e591a248a25b849612e448972068f10/server/src/main/java/org/elasticsearch/tasks/CancellableTask.java#L20
[TransportService]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/transport/TransportService.java
[Task management API]:https://www.elastic.co/guide/en/elasticsearch/reference/current/tasks.html
[cat task management API]:https://www.elastic.co/guide/en/elasticsearch/reference/current/cat-tasks.html
[TransportAction]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/action/support/TransportAction.java
[NodeClient#executeAndReturnTask]:https://github.com/elastic/elasticsearch/blob/5e8fd548b959039b6b77ad53715415b429568bc0/server/src/main/java/org/elasticsearch/client/internal/node/NodeClient.java#L100
[TaskManager#registerAndExecute]:https://github.com/elastic/elasticsearch/blob/5e8fd548b959039b6b77ad53715415b429568bc0/server/src/main/java/org/elasticsearch/tasks/TaskManager.java#L174
[RequestHandlerRegistry#processMessageReceived]:https://github.com/elastic/elasticsearch/blob/5e8fd548b959039b6b77ad53715415b429568bc0/server/src/main/java/org/elasticsearch/transport/RequestHandlerRegistry.java#L65

The tasks infrastructure is used to track currently executing operations in the Elasticsearch cluster. The [Task management API] provides an interface for querying, cancelling, and monitoring the status of tasks.

Each individual task is local to a node, but can be related to other tasks, on the same node or other nodes, via a parent-child relationship.

> [!NOTE]
> The Task management API is experimental/beta, its status and outstanding issues can be tracked [here](https://github.com/elastic/elasticsearch/issues/51628).

### Task tracking and registration

Tasks are tracked in-memory on each node in the node's [TaskManager], new tasks are registered via one of the [TaskManager#register] methods.
Registration of a task creates a [Task] instance with a unique-for-the-node numeric identifier, populates it with some metadata and stores it in the [TaskManager].

The [register][TaskManager#register] methods will return the registered [Task] instance, which can be used to interact with the task. The [Task] class is often sub-classed to include task-specific data and operations. Specific [Task] subclasses are created by overriding the [createTask][TaskAwareRequest#createTask] method on the [TaskAwareRequest] passed to the [TaskManager#register] methods.

When a task is completed, it must be unregistered via [TaskManager#unregister].

#### A note about task IDs
The IDs given to a task are numeric, supplied by a counter that starts at zero and increments over the life of the node process. So while they are unique in the individual node process, they would collide with IDs allocated after the node restarts, or IDs allocated on other nodes.

To better identify a task in the cluster scope, a tuple of persistent node ID and task ID is used. This is represented in code using the [TaskId] class and serialized as the string `{node-ID}:{local-task-ID}` (e.g. `oTUltX4IQMOUUVeiohTt8A:124`). While [TaskId] is safe to use to uniquely identify tasks _currently_ running in a cluster, it should be used with caution as it can collide with tasks that have run in the cluster in the past (i.e. tasks that ran prior to a cluster node restart).

Some long-running tasks are relocatable and can be handed off between nodes when the host node is shutting down, in which case the cluster-scope [TaskId] of the running task changes. To let callers correlate a task across hops, [TaskInfo] carries `originalTaskId` and `originalStartTimeMillis` fields that pin the original identity through the chain. See [Relocatable Tasks](#relocatable-tasks) for details.

### What Tasks Are Tracked

The purpose of tasks is to provide management and visibility of the cluster workload. There is some overhead involved in tracking a task, so they are best suited to tracking non-trivial and/or long-running operations. For smaller, more trivial operations, visibility is probably better implemented using telemetry APIs.

Some examples of operations that are tracked using tasks include:
- Execution of [TransportAction]s
  - [NodeClient#executeAndReturnTask] invokes [TaskManager#registerAndExecute]
  - [RequestHandlerRegistry#processMessageReceived] registers tasks for actions that are spawned to handle [TransportRequest]s
- Publication of cluster state updates

### Tracking a Task Across Threads and Nodes

#### ThreadContext

[ThreadContext]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java
[ThreadPool]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/threadpool/ThreadPool.java
[ExecutorService]:https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/ExecutorService.html

All [ThreadPool] threads have an associated [ThreadContext]. The [ThreadContext] contains a map of headers which carry information relevant to the operation currently being executed. For example, a thread spawned to handle a REST request will include the HTTP headers received in that request.

When threads submit work to an [ExecutorService] from the [ThreadPool], those spawned threads will inherit the [ThreadContext] of the thread that submitted them. When [TransportRequest]s are dispatched, the headers from the sending [ThreadContext] are included and then loaded into the [ThreadContext] of the thread handling the request. In these ways, [ThreadContext] is preserved across threads involved in an operation, both locally and on remote nodes.

#### Headers

[Task#HEADERS_TO_COPY]:https://github.com/elastic/elasticsearch/blob/5e8fd548b959039b6b77ad53715415b429568bc0/server/src/main/java/org/elasticsearch/tasks/Task.java#L62
[ActionPlugin#getTaskHeaders]:https://github.com/elastic/elasticsearch/blob/5e8fd548b959039b6b77ad53715415b429568bc0/server/src/main/java/org/elasticsearch/plugins/ActionPlugin.java#L99
[X-Opaque-Id API DOC]:https://www.elastic.co/guide/en/elasticsearch/reference/current/tasks.html#_identifying_running_tasks

When a task is registered by a thread, a subset (defined by [Task#HEADERS_TO_COPY] and any [ActionPlugin][ActionPlugin#getTaskHeaders]s loaded on the node) of the headers from the [ThreadContext] are copied into the [Task]'s set of headers.

One such header is `X-Opaque-Id`. This is a string that [can be submitted on REST requests][X-Opaque-Id API DOC], and it will be associated with all tasks created on all nodes in the course of handling that request.

#### Parent/child relationships

[ParentTaskAssigningClient]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/client/internal/ParentTaskAssigningClient.java
[TaskAwareRequest#setParentTask]:https://github.com/elastic/elasticsearch/blob/5e8fd548b959039b6b77ad53715415b429568bc0/server/src/main/java/org/elasticsearch/tasks/TaskAwareRequest.java#L20
[TransportService#sendChildRequest]:https://github.com/elastic/elasticsearch/blob/c47162afca78f7351e30accc4857fd4bb38552b7/server/src/main/java/org/elasticsearch/transport/TransportService.java#L932

Another way to track the operations of a task is by following the parent/child relationships. When registering a task it can be optionally associated with a parent task. Generally if an executing task initiates sub-tasks, the ID of the executing task will be set as the parent of any spawned tasks (see [ParentTaskAssigningClient], [TransportService#sendChildRequest] and [TaskAwareRequest#setParentTask] for how this is implemented for [TransportAction]s).

### Kill / Cancel A Task

[TaskManager#cancelTaskAndDescendants]:https://github.com/elastic/elasticsearch/blob/5e8fd548b959039b6b77ad53715415b429568bc0/server/src/main/java/org/elasticsearch/tasks/TaskManager.java#L811
[BanParentRequestHandler]:https://github.com/elastic/elasticsearch/blob/5e8fd548b959039b6b77ad53715415b429568bc0/server/src/main/java/org/elasticsearch/tasks/TaskCancellationService.java#L356
[UnregisterChildTransportResponseHandler]:https://github.com/elastic/elasticsearch/blob/5e8fd548b959039b6b77ad53715415b429568bc0/server/src/main/java/org/elasticsearch/transport/TransportService.java#L1763
[Cancel Task REST API]:https://www.elastic.co/guide/en/elasticsearch/reference/current/tasks.html#task-cancellation
[RestCancellableNodeClient]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/rest/action/RestCancellableNodeClient.java
[TaskCancelledException]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/tasks/TaskCancelledException.java

Some long-running tasks are implemented to be cancel-able. Cancellation of a task and its descendants can be done via the [Cancel Task REST API] or programmatically using [TaskManager#cancelTaskAndDescendants]. Perhaps the most common use of cancellation you will see is cancellation of [TransportAction]s dispatched from the REST layer when the client disconnects, to facilitate this we use the [RestCancellableNodeClient].

In order to support cancellation, the [Task] instance associated with the task must extend [CancellableTask]. It is the job of any workload tracked by a [CancellableTask] to periodically check whether it has been cancelled and, if so, finish early. We generally wait for the result of a cancelled task, so tasks can decide how they complete upon being cancelled, typically it's exceptionally with [TaskCancelledException].

When a [Task] extends [CancellableTask] the [TaskManager] keeps track of it and any child tasks that it spawns. When the task is cancelled, requests are sent to any nodes that have had child tasks submitted to them to ban the starting of any further children of that task, and any cancellable child tasks already running are themselves cancelled (see [BanParentRequestHandler]).

When a cancellable task dispatches child requests through the [TransportService], it registers a proxy response handler that will instruct the remote node to cancel that child and any lingering descendants in the event that it completes exceptionally (see [UnregisterChildTransportResponseHandler]). A typical use-case for this is when no response is received within the time-out, the sending node will cancel the remote action and complete with a timeout exception.

Cancellation of a [relocatable task](#relocatable-tasks) is more involved:
the [TaskId] addressed by the cancel request can refer to a task that has since been handed off to another node,
where the destination task carries on under a different local task id.
To handle this, [TransportCancelTasksAction] fans out to every node and double-broadcasts when the request matches a relocatable action,
deduplicating responses by `originalTaskId` (mirroring the list-tasks approach).
`BulkByPaginatedSearchTask` also rejects cancellation during the handoff window via `ensureCancellable` with a `503 Service Unavailable`,
so a cancel that races an in-flight handoff cannot be silently lost.

### Publishing Task Results

[TaskResult]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/tasks/TaskResult.java
[TaskResultsService]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/tasks/TaskResultsService.java
[CAT]:https://www.elastic.co/guide/en/elasticsearch/reference/current/cat.html
[ActionRequest]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/action/ActionRequest.java
[ActionRequest#getShouldStoreResult]:https://github.com/elastic/elasticsearch/blob/b633fe1ccb67f7dbf460cdc087eb60ae212a472a/server/src/main/java/org/elasticsearch/action/ActionRequest.java#L32
[TaskResultStoringActionListener]:https://github.com/elastic/elasticsearch/blob/b633fe1ccb67f7dbf460cdc087eb60ae212a472a/server/src/main/java/org/elasticsearch/action/support/TransportAction.java#L149

A list of tasks currently running in a cluster can be requested via the [Task management API], or the [cat task management API]. The former returns each task represented using [TaskResult], the latter returning a more compact [CAT] representation.

Some [ActionRequest]s allow the results of the actions they spawn to be stored upon completion for later retrieval. If [ActionRequest#getShouldStoreResult] returns true, a [TaskResultStoringActionListener] will be inserted into the chain of response listeners. [TaskResultStoringActionListener] serializes the [TaskResult] of the [TransportAction] and persists it in the `.tasks` index using the [TaskResultsService].

The [Task management API] also exposes an endpoint where a task ID can be specified, this form of the API will return currently running tasks, or completed tasks whose results were persisted. Note that although we use [TaskResult] to return task information from all the JSON APIs, the `error` or `response` fields will only ever be populated for stored tasks that are already completed.

[TaskResultsService] also exposes a `storeResultIfAbsent` variant that writes with `OpType.CREATE` and treats a version conflict as success.
This is used during reindex task relocation (see [Relocatable Tasks](#relocatable-tasks)),
so that the source and destination of a handoff can both safely attempt to persist the source-side result, with the destination node winning the race.

### Relocatable Tasks

[BulkByPaginatedSearchTask]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/index/reindex/BulkByPaginatedSearchTask.java
[LeaderBulkByPaginatedSearchTaskState]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/index/reindex/LeaderBulkByPaginatedSearchTaskState.java
[WorkerBulkByScrollTaskState]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/index/reindex/WorkerBulkByScrollTaskState.java
[ResumeInfo]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/index/reindex/ResumeInfo.java
[TaskInfo]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/tasks/TaskInfo.java
[TaskRelocatedException]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/index/reindex/TaskRelocatedException.java
[Reindexer]:https://github.com/elastic/elasticsearch/blob/main/modules/reindex/src/main/java/org/elasticsearch/reindex/Reindexer.java
[AbstractAsyncBulkByPaginatedSearchAction]:https://github.com/elastic/elasticsearch/blob/main/modules/reindex/src/main/java/org/elasticsearch/reindex/AbstractAsyncBulkByPaginatedSearchAction.java
[ShutdownPrepareService]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/node/ShutdownPrepareService.java
[ReindexRelocationNodePicker]:https://github.com/elastic/elasticsearch/blob/main/modules/reindex/src/main/java/org/elasticsearch/reindex/ReindexRelocationNodePicker.java
[TransportResumeReindexAction]:https://github.com/elastic/elasticsearch/blob/main/modules/reindex/src/main/java/org/elasticsearch/reindex/TransportResumeReindexAction.java
[TransportGetTaskAction]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/get/TransportGetTaskAction.java
[TransportListTasksAction]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/TransportListTasksAction.java
[TransportCancelTasksAction]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/cancel/TransportCancelTasksAction.java
[TransportRethrottleAction]:https://github.com/elastic/elasticsearch/blob/main/modules/reindex/src/main/java/org/elasticsearch/reindex/TransportRethrottleAction.java
[RestReindexAction]:https://github.com/elastic/elasticsearch/blob/main/modules/reindex/src/main/java/org/elasticsearch/reindex/RestReindexAction.java
[ReindexPlugin]:https://github.com/elastic/elasticsearch/blob/main/modules/reindex/src/main/java/org/elasticsearch/reindex/ReindexPlugin.java
[ReindexManagementPlugin]:https://github.com/elastic/elasticsearch/blob/main/modules/reindex-management/src/main/java/org/elasticsearch/reindex/management/ReindexManagementPlugin.java
[TransportCancelReindexAction]:https://github.com/elastic/elasticsearch/blob/main/modules/reindex-management/src/main/java/org/elasticsearch/reindex/management/TransportCancelReindexAction.java
[PitPaginatedHitSource]:https://github.com/elastic/elasticsearch/blob/main/modules/reindex/src/main/java/org/elasticsearch/reindex/PitPaginatedHitSource.java
[ClientPitPaginatedHitSource]:https://github.com/elastic/elasticsearch/blob/main/modules/reindex/src/main/java/org/elasticsearch/reindex/ClientPitPaginatedHitSource.java
[RemotePitPaginatedHitSource]:https://github.com/elastic/elasticsearch/blob/main/modules/reindex/src/main/java/org/elasticsearch/reindex/remote/RemotePitPaginatedHitSource.java
[ClientScrollablePaginatedHitSource]:https://github.com/elastic/elasticsearch/blob/main/modules/reindex/src/main/java/org/elasticsearch/reindex/ClientScrollablePaginatedHitSource.java
[RemoteScrollablePaginatedHitSource]:https://github.com/elastic/elasticsearch/blob/main/modules/reindex/src/main/java/org/elasticsearch/reindex/remote/RemoteScrollablePaginatedHitSource.java

Some long-running tasks can be _relocated_ from a node that is preparing to shut down onto another node,
where they resume from a saved checkpoint.
The destination task carries on under a new local task id but is identified to users as the same logical task.
Relocation is distinct from [Persistent Tasks](#persistent-tasks): there is no master-managed reassignment and no cluster state record of the task.
The source node decides directly which node to hand off to, sends a transport-level resume request, and the destination resumes from a pagination cursor passed inline.

Today the only action wired up for relocation is reindex. The state machine on [BulkByPaginatedSearchTask] and the `_tasks`-layer support is generic, so update-by-query and delete-by-query (which share `BulkByPaginatedSearchTask`) can be enabled with localized changes; see [Enabling relocation for another action](#enabling-relocation-for-another-action) below.

#### Feature gating and module layout

Relocation is gated by several flags:
- `ReindexPlugin#RELOCATE_ON_SHUTDOWN_NODE_FEATURE` (`reindex_relocate_on_shutdown`) is a cluster-level [`NodeFeature`] used by [RestReindexAction] to check that every node supports the protocol before opting an incoming request in.
- `ReindexPlugin#REINDEX_PIT_SEARCH_FEATURE` (`reindex_pit_search`) gates point-in-time-based pagination over using `scroll`.

The module layout reflects who owns what:
- `server` contains [BulkByPaginatedSearchTask], [ResumeInfo], [TaskInfo], [TaskRelocatedException], and the relocation-aware bits of [TransportGetTaskAction] and [TransportListTasksAction]. These do not depend on the reindex module.
- `modules/reindex` owns the execution and handoff: [Reindexer], [AbstractAsyncBulkByPaginatedSearchAction], [TransportResumeReindexAction], [ReindexRelocationNodePicker] (with stateful and stateless implementations), [TransportRethrottleAction], and the PIT/scroll pagination sources.
- [ReindexManagementPlugin] in `modules/reindex-management` owns the `_reindex/{id}` get/list/cancel/rethrottle endpoints. They delegate to the generic `_tasks` transport actions and overlay relocation identity on the response.

[`NodeFeature`]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/features/NodeFeature.java

#### Identity model across hops

A task's cluster-scope id is `{nodeId}:{localTaskId}` (see [Task IDs](#a-note-about-task-ids)).
When a task relocates, the host node id changes, so the [TaskId] of the running task changes.
To keep the user-facing identity stable, [TaskInfo] carries two extra fields:

- `originalTaskId`: the [TaskId] of the first task in the chain (i.e. the [TaskId] when the request was first received).
- `originalStartTimeMillis`: the start time of that first task, used by callers to render running time across hops.

Both fields propagate via [ResumeInfo.RelocationOrigin] across the wire.
`TaskInfo#withOriginalRelocationIdentity` is how `_reindex/{id}` and `_reindex` (list) endpoints overlay this identity on the latest task's [TaskInfo] for the response:
they keep the latest task's `taskId` (so the client can address the live task in subsequent calls) but reset `startTime`/`runningTime` to span the full chain.

[ResumeInfo.RelocationOrigin]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/index/reindex/ResumeInfo.java

#### Relocation flow

```mermaid
sequenceDiagram
    participant Shutdown as ShutdownPrepareService (source)
    participant Worker as BulkByPaginatedSearchTask (source)
    participant Reindexer as Reindexer (source)
    participant Picker as ReindexRelocationNodePicker
    participant Target as TransportResumeReindexAction (target)
    participant Tasks as ".tasks index"

    Shutdown->>Worker: maybeRequestRelocationForBulkByPaginatedSearch → requestRelocation()
    Worker->>Worker: notifyDone() builds ResumeInfo "(PIT or scroll id + status)"
    Worker->>Reindexer: complete with ResumeInfo
    Reindexer->>Picker: pick target node
    Reindexer->>Worker: tryInitiateRelocationHandoff() → HANDOFF_INITIATED
    Reindexer->>Target: ResumeReindexAction(ResumeInfo)
    Target->>Target: AbstractAsyncBulkByPaginatedSearchAction.start() resumes from PIT/scroll
    Target->>Tasks: storeRelocationSourceTaskResult "(upsert with TaskRelocatedException)"
    Worker->>Tasks: storeResultIfAbsent "(CREATE, loses race vs destination)"
```

Step by step:

1. [ShutdownPrepareService] reacts to the cluster shutdown metadata and, for each in-flight `BulkByPaginatedSearchTask`, calls `maybeRequestRelocationForBulkByPaginatedSearch` which invokes `BulkByPaginatedSearchTask#requestRelocation`. Tasks that are not `eligibleForRelocationOnShutdown` (e.g. synchronous requests where a client is blocking on the response) are skipped.
2. The worker in [AbstractAsyncBulkByPaginatedSearchAction]`#notifyDone()` observes that relocation has been requested at a batch boundary, builds a `ResumeInfo` containing either a PIT cursor or a scroll id together with the current status (counters, throttle, slice information), and calls `cleanupWithoutClosingPagination()` so that the search context (PIT or scroll) is not closed when the worker completes.
3. [Reindexer]`#listenerWithRelocations` receives the worker response, asks [ReindexRelocationNodePicker] for a target node (stateful and stateless variants live in `modules/reindex`), calls `BulkByPaginatedSearchTask#tryInitiateRelocationHandoff` to transition the `RelocationProgress` state machine to `HANDOFF_INITIATED`, and sends a `ResumeReindexAction` transport request to the target.
4. On the target, [TransportResumeReindexAction] feeds the `ResumeInfo` into `AbstractAsyncBulkByPaginatedSearchAction#start()`, which seeds the pagination source from the inherited PIT/scroll id and continues the run.
5. The destination calls `Reindexer#storeRelocationSourceTaskResult` to upsert a [TaskResult] for the _source_ task into `.tasks`. The stored result wraps a [TaskRelocatedException] that carries the destination [TaskId].
6. The source's own attempt to persist its [TaskResult] uses create-if-absent semantics (`BulkByPaginatedSearchTask#useCreateSemanticsForResultStorage` returns the `HANDOFF_INITIATED` flag), routed through [TaskManager]#storeTaskResult into `TaskResultsService#storeResultIfAbsent` (`OpType.CREATE`, version conflict treated as success). If the destination's upsert won the race, the source's write is a no-op.

#### Race conditions and current mitigations

There are four broad races between management APIs and an in-progress handoff:

- **List race.** A relocation can occur between the moment a list-tasks fan-out reaches the source and the moment it reaches the destination, so a task can appear in neither response. Mitigation: [TransportListTasksAction] performs a double-list and deduplicates results by `originalTaskId`, gated by `RELOCATABLE_ACTIONS`. For `wait_for_completion=true`, missing parents from the second pass are reconciled against `.tasks`. [AbstractAsyncBulkByPaginatedSearchAction]'s `relocationCooldownNanos` prevents back-to-back hops that would defeat dedup.
- **Cancel race.** A cancel addressed to the source [TaskId] may arrive after the task has relocated, so a naive cancel would either 404 on the source or complete "successfully" while the destination keeps running. Mitigation has two layers. First, [BulkByPaginatedSearchTask]'s `RelocationProgress` state machine (`NOT_STARTED` / `HANDOFF_INITIATED` / `TASK_CANCELLED`) ensures that if cancel races the handoff itself, one side loses on a CAS and the loser is rejected (the source returns `503 Service Unavailable` from `ensureCancellable` if the handoff has committed). Second, [TransportCancelTasksAction] is relocation-aware: for actions in `RELOCATABLE_ACTIONS` it always fans out to every node (`resolveNodes` ignores the target node id) and runs a double-broadcast, with `mergeResponses` deduplicating captured tasks by `originalTaskId` and dropping the stale 503/`ResourceNotFoundException` from the merge once the successor's cancel committed. The reindex-management `POST _reindex/{id}/_cancel` endpoint delegates to the same transport action so it inherits the behavior.
- **Rethrottle race.** Single-node-targeted rethrottle hits the same problem as cancel. Mitigation: [TransportRethrottleAction]`#followRelocationAndRethrottle`, gated on `RethrottleRequest#followRelocations` (set to true by `RestReindexRethrottleAction`), uses `GET _tasks/{id}` to follow the relocation chain to the current task, then re-issues the rethrottle. Total RPS is stored on the leader via `LeaderBulkByPaginatedSearchTaskState#setRequestsPerSecondWithRelocationGuard` and travels with `ResumeInfo` so the destination starts at the right rate.
- **Chain-break race.** If the source crashes before writing `.tasks`, or the `ResumeReindexAction` RPC is lost, the chain through `.tasks` can break and `GET _tasks/{id}` cannot follow the hop. Mitigation: the destination is responsible for upserting the source's `.tasks` entry, and the source writes with create-if-absent semantics. Either side can fail without orphaning the chain.

#### API relocation-awareness

The current state of each relevant endpoint:

- `GET _tasks/{id}` follows relocations via `TransportGetTaskAction#followReindexRelocationIfNeeded` (combined with `mergeRelocatedTask` and `extractRelocatedReindexTaskId`). Opt-out via `GetTaskRequest#setFollowRelocations`.
- `GET _tasks` (list) follows relocations via the double-list described above for any action in `RELOCATABLE_ACTIONS`.
- `GET _reindex/{id}` and `GET _reindex` delegate to the `_tasks` layer and overlay relocation identity on the response.
- `POST _tasks/{id}/_cancel` follows relocations for any action in `RELOCATABLE_ACTIONS` via [TransportCancelTasksAction]'s double-broadcast merge.
- `POST _reindex/{id}/_cancel` is relocation-aware by delegation: `TransportCancelReindexAction` issues a `TransportCancelTasksAction` request scoped to `ReindexAction.NAME`, and (for `wait_for_completion=true`) follows up with `TransportGetReindexAction` to materialise the final response with chain-aware identity.
- `POST _reindex/{id}/_rethrottle` follows relocations through [TransportRethrottleAction]. The response's `TaskInfo` entries pass through `TaskInfo#withOriginalRelocationIdentity` so the JSON shape matches the list-reindex output.

#### Sliced vs non-sliced

Relocation operates differently depending on whether the request was sliced:
- **Non-sliced**: a single worker holds the pagination cursor; on relocation it builds `ResumeInfo.WorkerResumeInfo` (with either `ScrollWorkerResumeInfo` or `PitWorkerResumeInfo`) and the leader for that single worker (which is the task itself) hands off.
- **Sliced**: each slice is its own worker task with a separate `WorkerBulkByScrollTaskState`. The leader's [LeaderBulkByPaginatedSearchTaskState] aggregates per-slice completion before initiating the relocation; the combined `ResumeInfo` carries one `SliceStatus` per slice and a `WorkerResumeInfo` for each unfinished slice. Already-finished slices propagate forward to ensure the final task on the last node sees a complete result. `BulkByPaginatedSearchParallelizationHelper` handles completed slices when resuming on the destination.

#### Pagination

Reindex's pagination has two implementations, abstracted by `PaginatedHitSource`:
- Scroll-based: [ClientScrollablePaginatedHitSource] (local) and [RemoteScrollablePaginatedHitSource] (remote).
- Point-in-time: [PitPaginatedHitSource] / [ClientPitPaginatedHitSource] / [RemotePitPaginatedHitSource], gated by `REINDEX_PIT_SEARCH_FEATURE`. Remote PIT requires the remote cluster to be at least 7.10; older remotes still fall back to scroll.

PIT is what makes relocation reliable: scroll contexts are bound to the search shard nodes that opened them, so on stateful clusters losing a search-shard node can invalidate the cursor mid-relocation. In stateless, PIT survives shard relocations and is backed by the object store. The TTL on whichever cursor is in use must outlive the handoff window: `AbstractAsyncBulkByPaginatedSearchAction#cleanupWithoutClosingPagination` keeps the context alive during the hop, and `ReindexSettings#REINDEX_PIT_KEEP_ALIVE_SETTING` controls the per-cluster PIT keep-alive used by reindex.

#### Enabling relocation for other BulkByPaginatedSearchTask actions

To opt update-by-query, delete-by-query, or any other [BulkByPaginatedSearchTask]-based action into the relocation protocol:

1. In the REST handler (e.g. `RestUpdateByQueryAction`), follow the pattern in [RestReindexAction]: behind a `clusterSupportsFeature.test(RELOCATE_ON_SHUTDOWN_NODE_FEATURE)` check and `wait_for_completion=false`, call `setEligibleForRelocationOnShutdown(true)` on the request. Synchronous requests should never be opted in, since the originating client is blocking on the response and cannot follow a hop.
2. Add the action name to `TransportListTasksAction#RELOCATABLE_ACTIONS` so the double-list / dedup path covers it. Without this, list responses can drop the task during a hop.
3. Add a resume transport action analogous to [TransportResumeReindexAction] (extending `AbstractResumeBulkByScrollAction`), and register it in the corresponding [ReindexPlugin]-style plugin. The execution side just needs to plumb `ResumeInfo` through `AbstractAsyncBulkByPaginatedSearchAction#start()`.
4. Ensure the request type propagates `eligibleForRelocationOnShutdown` to its slice subrequests (see `ReindexRequest#setEligibleForRelocationOnShutdown` and how it copies the flag onto the sliced request).
5. Update wire serialization: add the flag to the request's `Writeable` round-trip, and follow the project convention for adding a `TransportVersion` so old nodes do not see the field. See `AGENTS.md` for the `generateTransportVersion` workflow.
6. In [ShutdownPrepareService], extend the loop that calls `relocateReindexTasksAndAwaitComplete(...)` to also iterate the new action's tasks. The `maybeRequestRelocationForBulkByPaginatedSearch` helper already handles any `BulkByPaginatedSearchTask`, but the wait-for-completion driver in `awaitTasksComplete` is action-name scoped.
7. If you want a separate management surface (e.g. `_update_by_query/{id}` with relocation identity overlaid on responses), add a sibling to the `reindex-management` module. Otherwise, the generic `_tasks` APIs already cover the basics for any action listed in `RELOCATABLE_ACTIONS`.
8. Add integration tests modelled on `ReindexRelocationOnShutdownIT`, `ListTasksRelocationIT`, and `GetTaskRelocationIT`. Wire serialization tests should cover the new field; see `ReindexRequestWireSerializingTests`/`UpdateByQueryRequestWireSerializingTests`.

### Persistent Tasks

[PersistentTaskPlugin]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/plugins/PersistentTaskPlugin.java
[PersistentTasksExecutor]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/persistent/PersistentTasksExecutor.java
[PersistentTasksExecutor.Scope.Cluster]:https://github.com/elastic/elasticsearch/blob/f461731a30a6fe55d7d7b343d38426ddca1ac873/server/src/main/java/org/elasticsearch/persistent/PersistentTasksExecutor.java#L52
[PersistentTasksExecutor.Scope.Project]:https://github.com/elastic/elasticsearch/blob/f461731a30a6fe55d7d7b343d38426ddca1ac873/server/src/main/java/org/elasticsearch/persistent/PersistentTasksExecutor.java#L48
[PersistentTasksExecutorRegistry]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/persistent/PersistentTasksExecutorRegistry.java
[PersistentTasksNodeService]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/persistent/PersistentTasksNodeService.java
[PersistentTasksClusterService]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/persistent/PersistentTasksClusterService.java
[AllocatedPersistentTask]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/persistent/AllocatedPersistentTask.java
[ShardFollowTasksExecutor]:https://github.com/elastic/elasticsearch/blob/main/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutor.java
[HealthNodeTaskExecutor]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/health/node/selection/HealthNodeTaskExecutor.java
[SystemIndexMigrationExecutor]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/upgrades/SystemIndexMigrationExecutor.java
[PersistentTasksCustomMetadata]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/persistent/PersistentTasksCustomMetadata.java
[ClusterPersistentTasksCustomMetadata]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/persistent/ClusterPersistentTasksCustomMetadata.java
[PersistentTasksCustomMetadata.PersistentTask]:https://github.com/elastic/elasticsearch/blob/d466ad1c3c4cedc7d5f6ab5794abe7bfd72aef4e/server/src/main/java/org/elasticsearch/persistent/PersistentTasksCustomMetadata.java#L305

Up until now we have discussed only ephemeral tasks. If we want a task to survive node failures, it needs to be registered as a persistent task at the cluster level.

Plugins can register persistent tasks definitions by implementing [PersistentTaskPlugin] and returning one or more [PersistentTasksExecutor] instances. These are collated into a [PersistentTasksExecutorRegistry] which is provided to [PersistentTasksNodeService] active on each node in the cluster, and a [PersistentTasksClusterService] active on the master. A [PersistentTasksExecutor] can declare either [project][PersistentTasksExecutor.Scope.Project] or [cluster][PersistentTasksExecutor.Scope.Cluster] scope, but not both. A project scope task is not able to access data on a different project.

The [PersistentTasksClusterService] runs on the master to manage the set of running persistent tasks. It periodically checks that all persistent tasks are assigned to live nodes and handles the creation, completion, removal and updates-to-the-state of persistent task instances in the cluster state (see [PersistentTasksCustomMetadata] and [ClusterPersistentTasksCustomMetadata]).

The [PersistentTasksNodeService] monitors the cluster state to:
 - Start any tasks allocated to it (tracked in the local [TaskManager] by an [AllocatedPersistentTask])
 - Cancel any running tasks that have been removed ([AllocatedPersistentTask] extends [CancellableTask])

If a node leaves the cluster while it has a persistent task allocated to it, the master will re-allocate that task to a surviving node. To do this, it creates a new [PersistentTasksCustomMetadata.PersistentTask] entry with a higher `#allocationId`. The allocation ID is included any time the [PersistentTasksNodeService] communicates with the [PersistentTasksClusterService] about the task, it allows the [PersistentTasksClusterService] to ignore persistent task messages originating from stale allocations.

Some examples of the use of persistent tasks include:
 - [ShardFollowTasksExecutor]: Defined by [cross-cluster replication](#cross-cluster-replication-ccr) to poll a remote cluster for updates
 - [HealthNodeTaskExecutor]: Used to schedule work related to monitoring cluster health. This is currently the only example of a cluster scope persistent task.
 - [SystemIndexMigrationExecutor]: Manages the migration of system indices after an upgrade

### Integration with APM

[Traceable]:https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/telemetry/tracing/Traceable.java
[APM Spans]:https://www.elastic.co/guide/en/observability/current/apm-data-model-spans.html

Tasks are integrated with the ElasticSearch APM infrastructure. They implement the [Traceable] interface, and [spans][APM Spans] are published to represent the execution of each task.

# Cross Cluster Replication (CCR)

(Brief explanation of the use case for CCR)

(Explain how this works at a high level, and details of any significant components / ideas.)

### Cross Cluster Search

# Indexing / CRUD

[RestBulkAction]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/rest/action/document/RestBulkAction.java
[RestIndexAction]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/rest/action/document/RestIndexAction.java
[RestDeleteAction]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/rest/action/document/RestDeleteAction.java
[RestUpdateAction]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/rest/action/document/RestUpdateAction.java
[TransportBulkAction]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java
[TransportAbstractBulkAction]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/bulk/TransportAbstractBulkAction.java
[IngestActionForwarder]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/ingest/IngestActionForwarder.java
[BulkOperation]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/bulk/BulkOperation.java
[TransportShardBulkAction]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java
[TransportIndexAction]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java
[TransportDeleteAction]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/delete/TransportDeleteAction.java
[TransportUpdateAction]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/update/TransportUpdateAction.java
[TransportReplicationAction]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java
[TransportWriteAction]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportWriteAction.java
[IndexRequest]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/index/IndexRequest.java
[DeleteRequest]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/delete/DeleteRequest.java
[GlobalCheckpointSyncAction]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/seqno/GlobalCheckpointSyncAction.java
[UpdateRequest]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/update/UpdateRequest.java
[BulkRequest]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java
[IndexShardOperationPermits]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java
[SequenceNumbers]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/seqno/SequenceNumbers.java
[LocalCheckpointTracker]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/seqno/LocalCheckpointTracker.java
[ReplicationTracker]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/seqno/ReplicationTracker.java
[SeqNoFieldMapper]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/mapper/SeqNoFieldMapper.java
[ReplicationOperation]:https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java

The Distributed team owns Elasticsearch's write path. A typical write begins as an HTTP/REST request, is routed to
the appropriate primary shard, is applied to the shard's engine and translog, and is finally replicated across replicas
before an ack is sent back to the client.

Note that in serverless Elasticsearch, there is no replication process needed. The same reliability and fault tolerance
are instead achieved by uploading the translog to a blob store and relying on the store's durability and fault tolerance
guarantees. On an `index` node the `TranslogReplicator` batches operations into a node-level `NodeTranslogBuffer` and
uploads them to the object store, and the `StatelessCommitService` batches Lucene commits for upload (see the
[Stateless Roles](#stateless-roles) section for the class-level detail). See this
[blog post](https://www.elastic.co/search-labs/blog/thin-indexing-shards-elasticsearch-serverless) for more details.

The Distributed team also owns select parts of the read path (e.g. real-time `GET` requests targeting the
[translog](#translog)), but broader search capabilities like query execution, scoring, and aggregations fall under
the Search team.

This section follows a bulk index request end to end in stateful Elasticsearch, using [RestBulkAction]
as the starting point.

For a higher-level overview of the read and write paths, see
[Reading and writing documents](https://www.elastic.co/docs/deploy-manage/distributed-architecture/reading-and-writing-documents).

## The Write Path

### Coordinator: REST to Transport

A user sends a `POST` to `/_bulk` with a newline-delimited body of index, update, or delete actions
(see [documentation](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-bulk)).
The HTTP stack (described in [HTTP Server](#http-server)) hands the request to [RestBulkAction], which
[parses](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/rest/action/document/RestBulkAction.java#L120)
the body into a [BulkRequest] containing one [IndexRequest] (or [UpdateRequest] / [DeleteRequest]) per requested
operation. It then [executes](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/rest/action/document/RestBulkAction.java#L138)
a [TransportBulkAction] on the receiving node (see the [Transport](#transport) section for more details).

Single-document requests (`PUT` or `POST` to `/{index}/_doc/{id}`) follow the same path: [RestIndexAction]
builds an [IndexRequest] that [TransportIndexAction] wraps in a one-item [BulkRequest] before delegating to
[TransportBulkAction].

The node that received the request is the coordinating node for this request. On that node, [TransportBulkAction]
coordinates the rest of the write path.

Before routing to shards, [TransportAbstractBulkAction] (the parent class of [TransportBulkAction])
[resolves](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/bulk/TransportAbstractBulkAction.java#L277)
the ingest pipeline for each item by checking the request's `pipeline` parameter, the index's `default_pipeline`, and
the `final_pipeline` from the matching index template. If any item has a pipeline, the coordinating node either executes
the pipelines locally (if it is an ingest node, see the [Node Roles](#node-roles) section) or
[forwards](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/bulk/TransportAbstractBulkAction.java#L315)
the entire request to an ingest node via [IngestActionForwarder]. Once ingest finishes, each processed request has
its `pipeline` and `finalPipeline` fields
[reset](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/ingest/IngestService.java#L1092)
to `"_none"` and the request is
[re-submitted](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/bulk/TransportAbstractBulkAction.java#L364)
through `TransportAbstractBulkAction#applyPipelinesAndDoInternalExecute`. On this second pass `hasPipeline` is false
for every item and the ingest path is skipped.

[TransportBulkAction] also
[checks](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java#L327)
whether the target index or data stream exists, auto-creating missing indices via `AutoCreateAction` and
[rolling over](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java#L410)
data streams marked for lazy rollover before proceeding.

It then eventually runs a [BulkOperation] on the coordinator. `BulkOperation` is the class that resolves each item to
a concrete index and shard before sending work to primaries.

If the client did not supply an `_id`, one is
[generated](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/bulk/BulkOperation.java#L333)
before routing. The latest applied [ClusterState](#cluster-state) supplies routing for the chosen index (`routingTable`,
[ShardRouting](#cluster-state)). A routing key is chosen: by default the document `_id`, though the request may
[set](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-index) `routing` explicitly.
The routing key is then hashed to a shard number by `IndexRouting.hashToShardId`. For time series indices the key is
[derived from the document's dimension fields](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/routing/IndexRouting.java#L62).
For indices with a `routing_path` configured, it is derived from the configured value.

The coordinator now knows which shard holds the document. It still needs the current primary node for that shard.

Note that for a bulk request
([multiple documents in one HTTP call](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-bulk)),
[BulkOperation] [groups](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/bulk/BulkOperation.java#L194)
items by shard before dispatching each shard’s work to the appropriate primaries.

### Primary Routing & Execution

#### Primary Routing

Elasticsearch uses primary–backup replication. The primary shard copy defines the ordered write history and replicas
apply the same operations.

After matching each document to a shard, [BulkOperation] will
[hand over](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/bulk/BulkOperation.java#L485)
the rest of the execution to a [TransportShardBulkAction]. This is a subclass of [TransportReplicationAction], which
holds the core replication framework logic for document writes and deletes.

The coordinator uses the cluster state routing table ([Cluster State](#cluster-state)) to send the shard request to
the node that currently holds the primary.

In practice, the request can take more than one transport hop before it executes on the primary. The receiving node
re-samples cluster state, and if the active primary for that shard is assigned elsewhere, [TransportReplicationAction]
[forwards](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java#L1043)
the replication request to that node. That “chase the primary” step can repeat while routing metadata
converges. To avoid redirect loops between nodes whose cluster state versions differ, the request carries a
[routedBasedOnClusterVersion](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java#L1021)
field. Each forward updates it to the forwarding node’s cluster state version so the next receiver
[is on at least](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java#L898)
that version before it redirects again. Retries and awaiting cluster state updates are
[bounded by the request’s timeout](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java#L874).
On each retry attempt the node [checks](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java#L1102)
whether the timeout has expired, and if so fails the request.

Once the request
[reaches](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java#L964)
the node that actually hosts the primary, it will get
[wrapped](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java#L988)
in a [ConcreteShardRequest](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java#L1388)
that includes the shard’s primary term and target allocation id. That lets the primary and replicas refuse operations
that were built for a superseded primary generation.

#### Primary Execution

On the primary node, `TransportShardBulkAction`
[applies](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java#L441)
the shard's bulk items through [IndexShard], the single entry point for shard-level work
(see [IndexShard](#indexshard) in [Engine & Store](#engine--store) sections for more details).

The primary will first try to
[acquire](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java#L484)
an operation permit via [IndexShardOperationPermits]. Note that recovery and relocation are operations that can
intentionally grab all available permits with the goal of blocking in-flight writes. In that case, the incoming
operation is [queued](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java#L219)
and will resume once the block lifts. If the permit is successfully acquired, the primary will then validate and
[prepare](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/shard/IndexShard.java#L1026)
the operation: create the mapping (if not already existing), parse the source, etc. `IndexShard` will then hand over the
request to the engine (typically [InternalEngine], see [Engine](#engine) section for more details) that
[generates](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java#L1239)
the sequence number for the operation, [updates](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java#L1262)
Lucene via an `IndexWriter`, and then
[appends](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java#L1276)
the operation to the [Translog](#translog).

Note that the translog write happens after the Lucene update. The translog is primarily a durability and recovery log for
acknowledged operations, not a write-ahead log in the classic database sense. The translog entry type also depends on
the Lucene outcome. A successful Lucene apply writes the full operation, while a failure with an already-assigned seq_no
[writes a no-op](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java#L1278)
to preserve the sequence number history.

Note that updates and deletes use soft deletes in Lucene (tombstones and retention policies), not physical deletion,
and are therefore quite similar to writes.
See [Engine & Store](#engine--store) and [Segment Merges](#segment-merges) for more details.

### Replication

After the primary applies the operation, a [ReplicationOperation]
[wrapper object](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java#L592)
will coordinate the rest. It
[loads](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java#L151)
the replication group after the primary write (to avoid missing new recovery targets),
captures the global checkpoint and related metadata from the primary, then
[forwards the request](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java#L188)
to each replica in that group. Each replica
[applies](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java#L1331)
the same operation (using the same `seq_no` and primary term) to its engine and returns checkpoint information.

Allocation ids that are still in the in-sync set but no longer have a live shard in routing are
[marked stale](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java#L187)
on the primary. The master will update the cluster state to drop those allocation ids from the index’s
[in-sync set](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/metadata/IndexMetadata.java#L651)
before the write is acknowledged back to the client. Similarly, if a replica fails to execute the replicated operation,
it will also get dropped from the in-sync set before the write is acknowledged back. The primary will not
acknowledge the replication operation as “done” while the cluster state still reports an in-sync copy that has failed
the operation.
[A ref-counting listener](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java#L108)
also waits for the primary shard application, each replica response, and post-replication hooks. The client only gets an
ack after that coordinated completion, together with `wait_for_active_shards` and translog / `syncAfterWrite` behaviour (see
[Translog](#translog) and `TransportWriteAction`).

```mermaid
sequenceDiagram
    participant C as Client
    participant CN as Coordinating node
    participant P as Primary shard copy
    participant R as Replica shard copy
    participant M as Elected master

    C->>CN: REST index (RestIndexAction → IndexRequest)
    CN->>CN: Bulk coordinator: route to shard
    CN->>P: Shard bulk on primary (TransportShardBulkAction)
    P->>P: IndexShard / engine (Translog + Lucene)
    Note over P: ReplicationOperation: stale in-sync ids, then replica fan-out
    P->>R: Replica request (seq_no, primary_term, checkpoints)
    alt replica applies successfully
        R->>R: Engine apply (Translog + Lucene)
        R-->>P: ReplicaResponse (checkpoints)
    else replica fails replication
        R-->>P: failure
        P->>M: remoteShardFailed (shard-failed -> update in-sync / routing)
        M->>M: Cluster state task (drop failed copy from in-sync set, etc.)
        M-->>P: shard-failed applied (listener completes)
    end
    P-->>CN: Primary result + shard info (after all pending replica / master / post steps)
    CN-->>C: HTTP response (may include shard failures in body)
```

When replication and the ref-counted coordination above have finished, the primary completes the shard-level request and
returns the result to the node that issued that shard-level transport request. For a normal REST bulk request, that
issuer is the HTTP request coordinating node, which then completes the write action and sends
the HTTP response to the client. The same shard-level transport requests can also be generated internally, for example
via [TriggeredWatchStore#putAll](https://github.com/elastic/elasticsearch/blob/v9.3.0/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/execution/TriggeredWatchStore.java#L84)
in Watcher. The issuer is then whichever node runs that internal client, and the primary returns the result to that
transport caller.

### Primary Terms & Sequence Numbers

Every successful index, update, or delete operation on a shard is tagged with a sequence number (`seq_no`) and a primary term.
Together they identify that logical change for replication, recovery, and optimistic concurrency. Sentinel values and
helpers are defined on [SequenceNumbers] (for example `UNASSIGNED_SEQ_NO` on the way into the primary, and
`UNASSIGNED_PRIMARY_TERM` before a shard is operational).

The primary term is a monotonically increasing counter per shard, recorded in cluster state metadata (see
[IndexMetadata#primaryTerm](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/cluster/metadata/IndexMetadata.java#L1199)).
It advances when a new primary must take over following a primary failure, e.g. when a replica is promoted to primary
because the current primary crashed or became unavailable. Note that the primary term is not increased during graceful
primary relocation. The shard stamps all operations with the
[current term](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/shard/IndexShard.java#L514).

The sequence number is another monotonically increasing id for mutations on the shard history. It is
[computed](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java#L1163)
by the primary's [InternalEngine] before it applies an operation locally. Replicas apply the same `seq_no` supplied
by the primary. Their engine
[records](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java#L1256)
that sequence number as "seen" before indexing. After the operation, the local checkpoint also gets updated (see the following [Checkpoints](#checkpoints--gaps) subsection).

Both values are stored in Lucene for each live document. [SeqNoFieldMapper] indexes `_seq_no` as a numeric field (for
search and sort) and stores the primary term in doc values. If two copies
ever share the same `_seq_no` (for example after a primary promotion), the copy with the higher primary term wins.
Clients can also supply `if_seq_no` / `if_primary_term` on write requests (see
[IndexRequest#setIfSeqNo](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/index/IndexRequest.java#L623)),
in which case the engine will use optimistic concurrency control against the last known sequence number.

```mermaid
flowchart LR
    subgraph prim["Primary shard"]
        MD[IndexMetadata.primaryTerm in cluster state]
        TR[ReplicationTracker operation term]
        LCT[LocalCheckpointTracker next seq_no]
        ENG[InternalEngine assigns seq_no + term]
        MD --> TR
        TR --> ENG
        LCT --> ENG
    end
    ENG --> TX[ConcreteShardRequest / replica transport]
    TX --> REP[Replica engine applies fixed seq_no + term]
```

### Checkpoints & Gaps

Each shard copy tracks how far it has applied the shared history of `seq_no` values using a local checkpoint,
which is the highest sequence number for which that copy has processed every earlier `seq_no` (inclusive).
The [InternalEngine] holds a [LocalCheckpointTracker]
[field](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java#L172)
that maintains that marker and a separate persisted checkpoint for what is durably on disk.

The primary also tracks a
[global checkpoint](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/seqno/ReplicationTracker.java#L147),
which is the sequence number up to which all in-sync copies have
processed every earlier operation. The [ReplicationTracker] computes it as the minimum of those per-copy local checkpoints among in-sync shard copies.
The primary [updates](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/seqno/ReplicationTracker.java#L1375)
the global checkpoint whenever an in-sync copy’s local checkpoint advances, when the primary activates or takes over
after relocation, when a copy becomes in-sync, or when cluster state drops tracked copies.
During operation replication,
[replica responses carry](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java#L290)
local and global checkpoint fields so the primary can [derive](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/seqno/ReplicationTracker.java#L1320)
peer progress. After a successful write, [TransportReplicationAction] will also
[trigger](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java#L568)
a [GlobalCheckpointSyncAction] when there are no more operations in-flight (`maxSeqNo == globalCheckpoint`).
Indeed, the global checkpoint is piggybacked on every replication operation, but once the last operation completes,
the checkpoint may advance further with no subsequent operation to carry it to the replicas.

The global checkpoint is the replication-group safety line: it is the highest `seq_no` known to be processed
on every in-sync copy, so it underpins
[retention leases](https://www.elastic.co/docs/reference/elasticsearch/index-settings/history-retention),
[translog](#translog) truncation, [soft-delete retention](#segment-merges), and [peer recovery](#peer-recovery)
coordination.

Because replication fan-out is concurrent, a replica may receive operations with out-of-order `seq_no` values.
The engine still applies each operation once its prerequisites are satisfied. The `LocalCheckpointTracker`
[records](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java#L1299)
which sequence numbers have been processed (including pending gaps) so the local checkpoint only advances when
the [contiguous prefix is complete](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/seqno/LocalCheckpointTracker.java#L194).
To save memory, `LocalCheckpointTracker` tracks which `seq_no` values are still pending
[using sparse bit sets](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/seqno/LocalCheckpointTracker.java#L32)
in chunk-keyed maps.

Replicas cache the global checkpoint value the primary advertises and refresh it from ongoing replication traffic
(for example, each [ConcreteReplicaRequest](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java#L1525)
carries the primary’s global checkpoint for the replica permit path,
and each [ReplicaResponse](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/action/support/replication/TransportReplicationAction.java#L1284)
returns the replica’s local checkpoint and last synced global checkpoint).

Note that the `InternalEngine` also tracks the
[maxSeqNoOfUpdatesOrDeletes](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/Engine.java#L2512)
(MSU) for write path optimization purposes. The [Engine](#engine)
[javadoc](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/Engine.java#L2479)
goes into details about the `LCP < MSU` versus `MSU <= LCP` cases.

The engine also applies an append-only
[optimization](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java#L1328)
for auto-generated `_id`s. When a document has an auto-generated ID and the operation is not flagged as a retry,
the engine will try to
[assert](https://github.com/elastic/elasticsearch/blob/v9.3.0/server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java#L1513)
that no concurrent retry with an equal or higher timestamp has landed on the shard so it can skip the version map
lookup entirely and call `IndexWriter.addDocument` instead of `updateDocument`. If the check fails, the engine falls
back to `updateDocument` to guard against creating a duplicate.

# Server Startup

# Server Shutdown

### Closing a Shard

(this can also happen during shard reallocation, right? This might be a standalone topic, or need another section about it in allocation?...)

# Shrink/Split/Clone index APIs

These APIs are used to create a new index that contains a copy of data from a provided index and differs in number of shards and/or index settings. They can only be executed against source indices that are marked read-only.

The shrink API (`/{index}/_shrink/{target}`) creates an index that has fewer shards than the original index.

The split API (`/{index}/_split/{target}`) creates an index that has more shards than the original index.

The clone API (`/{index}/_clone/{target}`) creates an index that has the same number of shards as the original index but may have different index settings.

The main implementation logic is centralized in `TransportResizeAction` however it only creates a new index in the cluster state using special `recoverFrom` and `resizeType` parameters. The entire workflow involves multiple components.

The high level structure is the following:
1. `TransportResizeAction` creates index metadata for the new index that contains information about the resize performed. It also creates a routing table for the new index which assigns a `LocalShardsRecoverySource.INSTANCE` recovery source to primary shards based on the resize information in the index metadata .
2. Allocation logic allocates shards of the index so that they are on the same node as the corresponding shards of the source index. See `ResizeAllocationDecider`. Note that this doesn't work for shrink case since during shrink there are multiple source shards that are "merged" together. These source shards may be on different nodes already. Shard movement in this case needs to be performed manually.
3. Primary shards perform [local shards recovery](#local-shards-recovery) using index metadata to know what type of resize operation is performed.

# Health API

The Health API (`GET /_health_report`) provides a structured health report for the cluster. It is indicator-based: each health indicator evaluates a specific aspect of cluster health (for example shard allocation or disk) and returns a status (`GREEN`, `YELLOW`, `RED`, or `UNKNOWN`) plus structured details, impacts, and diagnoses. The top-level status is derived from the worst indicator status.

## Health node and data flow

A health node is selected by the master node via a persistent task. The health node maintains a cache of per-node health information. Each node periodically publishes its local health info to the health node cache. That cached health info is then used as input to health indicators when evaluating the overall cluster health. If the health node fails or leaves the cluster, the master node selects a new health node.

```
Publish: LocalHealthMonitor (local node) -> UpdateHealthInfoCacheAction -> HealthInfoCache (health node)
Retrieve: Health API (coordination node) -> FetchHealthInfoCacheAction -> HealthInfoCache (health node)
```

Relevant classes:
- [HealthNodeTaskExecutor][HealthNodeTaskExecutor]: the persistent task executor that selects the health node.
- [HealthInfoCache][HealthInfoCache]: contains the cached per-node health info on the health node.
- [LocalHealthMonitor][LocalHealthMonitor]: monitors local health and publishes to the health node.
- [HealthTracker][HealthTracker]: invoked by LocalHealthMonitor and tracks local health info on each node.

## Health indicators

There are two kinds of health indicators: preflight and regular. Preflight indicators run first; if any preflight indicator is not GREEN, regular indicators return UNKNOWN to avoid misleading results on an unstable cluster. Currently, the only preflight indicator is [StableMasterHealthIndicatorService][StableMasterHealthIndicatorService], which checks whether the cluster has a stable master node.

Relevant classes:
- [HealthIndicatorService][HealthIndicatorService]: the interface for health indicators.
- [HealthService][HealthService]: orchestrates indicator evaluation.

## Health periodic logger

Cluster health status can be logged periodically by [HealthPeriodicLogger][HealthPeriodicLogger], which runs on the health node and calls [HealthService][HealthService] at a configured interval. It enables alerting on the health status.

## Health metadata

Some health configuration is stored in the cluster state [HealthMetadata][HealthMetadata]. It stores user-configurable health settings, such as disk watermark thresholds. The current master node publishes its settings to the cluster state so the same settings are used across nodes in the cluster.

Relevant classes:
[HealthMetadataService][HealthMetadataService]: runs on master node, publishes health metadata to the cluster state.

[HealthService]: https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/health/HealthService.java
[HealthIndicatorService]: https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/health/HealthIndicatorService.java
[HealthInfoCache]: https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/health/node/HealthInfoCache.java
[LocalHealthMonitor]: https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/health/node/LocalHealthMonitor.java
[HealthPeriodicLogger]: https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/health/HealthPeriodicLogger.java
[HealthTracker]: https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/health/node/tracker/HealthTracker.java
[StableMasterHealthIndicatorService]: https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/cluster/coordination/StableMasterHealthIndicatorService.java
[HealthMetadata]: https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/health/metadata/HealthMetadata.java
[HealthMetadataService]: https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/health/metadata/HealthMetadataService.java

# Watcher

[Watcher]: https://github.com/elastic/elasticsearch/blob/main/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/Watcher.java
[WatcherLifeCycleService]: https://github.com/elastic/elasticsearch/blob/main/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/WatcherLifeCycleService.java
[ExecutionService]: https://github.com/elastic/elasticsearch/blob/main/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/execution/ExecutionService.java
[WatcherService]: https://github.com/elastic/elasticsearch/blob/main/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/WatcherService.java
[TickerScheduleTriggerEngine]: https://github.com/elastic/elasticsearch/blob/main/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/trigger/schedule/engine/TickerScheduleTriggerEngine.java
[EmailAction]: https://github.com/elastic/elasticsearch/blob/main/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/actions/email/EmailAction.java
[EmailService]: https://github.com/elastic/elasticsearch/blob/main/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/notification/email/EmailService.java
[WebhookAction]: https://github.com/elastic/elasticsearch/blob/main/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/actions/webhook/WebhookAction.java
[WebhookService]: https://github.com/elastic/elasticsearch/blob/main/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/notification/WebhookService.java
[Actions Package]: https://github.com/elastic/elasticsearch/blob/main/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/actions
[ReportingAttachment]: https://github.com/elastic/elasticsearch/blob/main/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/notification/email/attachment/ReportingAttachment.java

Watcher lets you set a schedule to run a query, and if a condition is met it executes an action.
As an example, the following performs a search every 10 minutes. If the number of hits found is greater than 0 then it logs an error message.
```console
PUT _watcher/watch/log_error_watch
{
  "trigger" : { "schedule" : { "interval" : "10m" }},
  "input" : {
    "search" : {
      "request" : {
        "indices" : [ "logs" ],
        "body" : {
          "query" : {
            "match" : { "message": "error" }
          }
        }
      }
    }
  },
  "condition" : {
    "compare" : { "ctx.payload.hits.total" : { "gt" : 0 }}
  },
  "actions" : {
    "log_error" : {
      "logging" : {
        "text" : "Found {{ctx.payload.hits.total}} errors in the logs"
      }
    }
  }
}

```

## How Watcher Works

- We have an API to define a “watch”, which includes the schedule, the query, the condition, and the action
- Watch definitions are kept in the `.watches` index
- Information about currently running watches is in the `.triggered_watches` index
- History is written to the `.watcher_history` index
- Watcher ([WatcherLifeCycleService]) runs on all nodes, but only executes watches on a node that has a copy of the `.watches` shard that the particular watch is in (see [WatcherService])
  -  Uses a hash to choose the node if there is more than one shard
- Example common use cases:
  - Periodically send data to a 3rd party system
  - Email users with alerts if certain conditions appear in log files
  - Periodically generate a report using Kibana, and email that report as an attachment. This is supported by declaring a [ReportingAttachment] [ReportingAttachment] to the [EmailAction] [EmailAction] in the watch definition.

## Relevant classes:

- [Watcher] [Watcher] – the plugin class
- [WatcherLifeCycleService] [WatcherLifeCycleService]  – created by the Watch plugin on each node
- [WatcherService] [WatcherService] – decides which watches this node ought to run
- [ExecutionService] [ExecutionService] – executes watches
- [TickerScheduleTriggerEngine] [TickerScheduleTriggerEngine] – handles the periodic (non-cron) schedules that we see the most
- [EmailAction] [EmailAction] / [EmailService] [EmailService] – emails to third-party email server
- [WebhookAction] [WebhookAction] / [WebhookService] [WebhookService] – sends requests to external endpoints
- [Various other actions] [Actions Package] (for example posting to Slack, Jira, etc.)

## Debugging

- The most useful debugging information is in the Elasticsearch logs and the `.watcher_history` index
- It is often useful to get the contents of the `.watches` index
- Frequent sources of problems:
  - There is no guarantee that an interval schedule watch will run at exactly the requested interval after the last run
  - In older versions (before 8.17), the counter for the interval schedule restarts if the shard moves. For example, if the interval is once every 12 hours, and the shard moves 10 hours into that interval, it will be at least 12 more hours until it runs.
  - Calls to remote systems ([EmailAction] and [WebhookAction]) are a frequent source of failures. Watcher sends the request but doesn't know what happens after that. If you see that the call was successful in `.watcher_history`, the best way to continue the investigation is in the logs of the remote system.
  - Even if watcher fails during a call to a remote system, the error is likely to be outside of watcher (e.g. network problems). Check the error message in `.watcher_history`.