---
metadata:
  - name: generator
    content: Diplodoc Platform v5.63.3
alternate:
  - https://ytsaurus.tech/docs/en/flow/concepts/guarantees.md
  - https://ytsaurus.tech/docs/ru/flow/concepts/guarantees.md
  - href: https://ytsaurus.tech/docs/en/flow/concepts/guarantees.md
    type: text/markdown
    title: Markdown version
  - href: https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65
    rel: describedby
---
> **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65

<!-- source: en/_includes/flow/concepts/guarantees.md -->
# Processing guarantees in YTsaurus Flow

## Overview {#overview}

Flow provides **exactly-once semantics** for event processing by default. This means:

- Consumers get the result of processing each input message exactly once, with no losses or duplicates.
- The [state](https://ytsaurus.tech/docs/en/flow/concepts/stateful.md) is updated atomically along with message processing.

If needed, you can relax the semantics to [**at-least-once**](#at-least-once) or [**at-most-once**](#at-most-once) — for example, to reduce latency or lower the pipeline’s resource consumption.

## Exactly-once semantics {#exactly-once}

### How it works {#how-it-works}

Exactly-once is ensured by three mechanisms that work together.

#### 1. Lease transaction {#lease}

A `Lease` is a controller-owned master transaction that the controller creates for each [job](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#job). Users do not create the Lease or control its lifetime. The controller keeps the Lease active with periodic pings and checks the state of all Leases on every scheduling cycle.

Every ordinary transaction in which a job commits [epoch](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#epoch) data lists that job’s Lease as a prerequisite in `prerequisite_transaction_ids`. The Lease must remain active for the commit to succeed. A Lease ends in one of two ways. The controller aborts it whenever it removes the job: during partition rebalancing, after a job failure or a lost worker, or when the pipeline stops, pauses, or completes. A Lease also expires on its own if the controller stops keeping it alive with pings. Both routes activate the same barrier.

After the Lease is aborted or expires, commits guarded by that prerequisite are rejected, so a zombie job cannot write data after the controller no longer considers it the owner of its [partition](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#partition). Even when it has no data, a job periodically commits a nearly empty transaction with the same prerequisite to detect that its Lease has been lost.

When a Lease expires, the controller notices it on the next scheduling cycle and removes the job it belonged to. While the pipeline remains active and scheduling continues, after a Lease is aborted or expires the controller removes the old job if it is still present, schedules the necessary replacement jobs for the affected current partitions, and creates a new Lease for each replacement job. Each replacement restores state from the last committed epoch, resumes processing the affected current partitions, and may reprocess only uncommitted work. When the pipeline completes, stops, or pauses, the controller instead removes jobs, terminates their Leases, and schedules no replacements.

Lease is a commit-time ownership barrier. By itself, it does not perform [input message deduplication](#input-dedup), provide [output data atomicity](#output-atomicity) or [state atomicity](#automatic-guarantees), or cover [arbitrary external side effects](#side-effects).

#### 2. Input message deduplication {#input-dedup}

- **Internal streams** (`input`): deduplication by `message_id` using the internal `input_messages` table or its compact variant `compact_input_messages`. The compact variant is selected automatically when `experimental_enable_non_uint_key` is disabled (partitioning is done exclusively by a `uint64` hash column); you can override this behavior with the `use_compact_input_messages` parameter. Older `message_id` values are removed from this table using [SystemWatermark](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#timestamps-and-watermarks).
- **Sources** (`source`): deduplication by offsets — consumer offsets advance only after a successful [epoch](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#epoch) commit.
- **[Timers](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#timer)**: the result of timer processing is committed in the same transaction as the timer’s deletion.

#### 3. Output data atomicity {#output-atomicity}

The way output messages are delivered depends on the [sink](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#sink) type:

- **Synchronous sinks** (for example, `TSyncQueueSink`) write data to the target system in the same epoch transaction. The write and acknowledgment are atomic.
- **Asynchronous sinks** (for example, `TAsyncQueueSink`) synchronously store messages in YTsaurus in the `output_messages` table, then send messages to the target system after the epoch is committed, using the `(producer_id, seqNo)` pair for deduplication on the receiver side (Queue API). If sending fails, messages from `output_messages` are sent again with the same `seqNo`. This essentially implements the [transactional outbox](https://en.wikipedia.org/wiki/Inbox_and_outbox_pattern#The_outbox_pattern) pattern with deduplication in the target system, if it supports it.

### What the developer must do {#developer-responsibilities}

Flow’s exactly-once guarantees apply to the pipeline’s internal state and built-in delivery mechanisms. However, your custom code can break these guarantees if you don’t follow these rules.

#### Computation determinism {#determinism}

Determinism requirements depend on the computation type.

- **For [Swift](https://ytsaurus.tech/docs/en/flow/concepts/swift.md) computations** (`TSwiftMapComputation`, `TSwiftOrderedSourceComputation`) determinism is strictly required. In Swift computations, computation results aren’t materialized between epochs and can be recalculated on restart. If different attempts to compute the same message yield different results, you might mix results from different attempts, or lose or duplicate data.
- **For regular computations** (`TTransformComputation`, `TTransformOrderedSourceComputation`) non-deterministic code is technically allowed: exactly one attempt (out of all attempts caused by job restarts) is applied atomically. No result mixing occurs. However, if your code produces different output on a rerun, it can lead to unexpected behavior — especially if it has side effects (like writing to external systems) or updates state based on non-deterministic values.

{% note warning "What breaks determinism" %}

- Using the current time (`Now()`, `time.time()`) in business logic.
- Generating random numbers.
- Calling external services whose results can change between calls.

{% endnote %}

#### Side effects {#side-effects}

Any action outside Flow — an HTTP request, writing to a file, sending a notification — **isn’t covered** by exactly-once guarantees. If a job restarts, such an action might be executed again.

If side effects are unavoidable, you must ensure their idempotency yourself (for example, by using the message’s unique ID as an idempotency key).

#### External state {#external-state}

Updates to Flow’s [internal state](https://ytsaurus.tech/docs/en/flow/concepts/stateful.md) (YSON-state, External State via `IExternalStateManager`) are atomic and safe. However, if a computation modifies data in an external system (a database, cache), those changes aren’t part of the epoch transaction and might be duplicated.

### What Flow guarantees automatically {#automatic-guarantees}

- Internal streams between [computations](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#computation) — exactly-once.
- [State](https://ytsaurus.tech/docs/en/flow/concepts/stateful.md) updates — atomic with message processing.
- Timer processing and deletion — in the same transaction.
- Advancing [source](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#source) offsets — only after a successful epoch commit.

## At-least-once semantics {#at-least-once}

At-least-once guarantees that each message is processed **at least once**, but allows repeated processing. This can be useful when your business logic is naturally idempotent and the cost of deduplication is unreasonably high.

### Mode `processing_mode: at_least_once_consistent` {#processing-mode}

You set the `processing_mode` parameter at the `TTransformComputation` level in the spec. The default value is `exactly_once`.

```
"computations" = {
    "MyComputation" = {
        "processing_mode" = "at_least_once_consistent";
        ...
    };
};
```

**What changes**: input message deduplication by `message_id` is disabled. Messages might be processed again after a job restart.

**What doesn’t change**: state remains consistent (that’s why “consistent” is in the mode name). Epoch transactionality, the Lease mechanism, and the atomicity of state updates all keep working.

**When to use it**: when deduplication overhead is significant and your business logic is idempotent (for example, `insert_or_assign` instead of `increment`).

### Batching in Swift computations: `allow_batching_with_relaxed_guarantees` {#swift-allow-batching-with-relaxed-guarantees}

You set the `allow_batching_with_relaxed_guarantees` parameter in the `parameters` block of a `TSwiftMapComputation` computation. The default value is `%false`.

```
"computations" = {
    "MyBatcher" = {
        "parameters" = {
            "allow_batching_with_relaxed_guarantees" = %true;
        };
        ...
    };
};
```

**What changes**: a single output message can combine multiple input messages (batching) — for example, to collapse many small messages into one large message and reduce the load on the number of messages per partition downstream. The combined message gets a `MessageId` derived deterministically from the set of its parents' `MessageId`s, and an input message is considered processed only after all its children are delivered.

**What this means for guarantees**: on restart, the batch boundaries might differ. A replay that reproduces a batch with the same set of parents yields the same `MessageId` and is deduplicated; but if a parent ends up in a batch with a different composition, that batch's `MessageId` is different — downstream computations must be ready to see each parent’s content more than once (at-least-once). Also, the `MessageId` order within a single key isn’t preserved on the downstream computations’ side: any logic that relies on `MessageId` ordering within a key must be rewritten to account for this.

**When to use it**: when you need to aggregate the message stream (batching/collapsing), and the downstream logic is idempotent and doesn’t depend on `MessageId` order within a key. With the flag disabled (default), each output message has exactly one parent and the Swift computation keeps exactly-once guarantees.


## At-most-once semantics {#at-most-once}

At-most-once guarantees that each message is processed **no more than once**, but allows message loss.

You achieve this semantics by configuring `at_most_once_strategy` on asynchronous sinks that support it.

```
"sinks" = {
    "my_sink" = {
        "at_most_once_strategy" = {
            "enabled" = %true;
            "total_queue_bytes_limit" = 104857600;  // 100 MB
            "suspend_destruction_duration" = 60000;   // ms
        };
        ...
    };
};
```

**How it works**: messages are placed in an internal queue. If the queue is full (`total_queue_bytes_limit`), new messages **are dropped without an error**. Deduplication and delivery guarantees are absent.

**When to use it**: for telemetry, metrics, best-effort notifications — scenarios where losing some messages is acceptable and throughput is more important than reliability.

## Ordering guarantees {#ordering}

Flow **doesn’t guarantee** global message processing order. However, there are guarantees for derived messages with matching [keys](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#key) within a single [lineage](https://ytsaurus.tech/docs/en/flow/concepts/lineage.md) chain.

For details, see the [Message processing order](https://ytsaurus.tech/docs/en/flow/concepts/ordering.md) section.

## How connectors affect guarantees {#connectors}

Each [connector](https://ytsaurus.tech/docs/en/flow/connectors/about.md) has its own set of guarantees, which depend on the connection type (source/sink) and the sink variant (sync/async/at-least-once).

### Queue (QYT) {#queue-guarantees}

- **Source**: exactly-once — consumer offsets advance only after epoch commit.
- **Synchronous sink** (`TSyncQueueSink`): exactly-once — writing to the queue happens in the main epoch transaction. Works only on the main processing cluster.
- **Asynchronous sink** (`TAsyncQueueSink`): exactly-once — messages are stored in `output_messages`, then delivered with `producer_id` + `seqNo`. Queue API deduplicates repeats. Works cross-cluster.

For more details, see [Queue](https://ytsaurus.tech/docs/en/flow/connectors/queue.md).

### Static Table {#static-table-guarantees}

- **Source**: exactly-once — deduplication by read ranges.
- **Sink** (`TArrivalOrderTableSink`): exactly-once — the output table and its progress commit in a single master transaction; a per-partition frontier deduplicates replay, so a partially covered replay writes only the uncovered tail without restarting the job; the delivery callback fires only after that external commit and the following Flow commit.

For more details, see [Static Table](https://ytsaurus.tech/docs/en/flow/connectors/static-table.md).

### HTTP {#http-guarantees}

- **Source**: none.
- **Asynchronous sink** (`TAsyncHttpSink`): at-least-once — POST requests run outside the epoch transaction, so the receiver must make the operation idempotent or deduplicate the stable message ID sent in `Idempotency-Key` by default. Enabling `at_most_once_strategy` makes spec loading fail.

For more details, see [HTTP extension](https://ytsaurus.tech/docs/en/flow/extensions/http.md).

### Kafka {#kafka-guarantees}

- **Source**: exactly-once — offsets are stored in the pipeline state and advance only after epoch commit.
- **Asynchronous sink** (`TKafkaSink`): exactly-once by default — messages are stored in `output_messages` and written after epoch commit in Kafka transactions together with a progress marker, and a restarted sink skips the messages the marker covers. Consumers must read with `isolation.level = read_committed`. If the marker is lost while messages may already be committed, the sink stops instead of writing possible duplicates. With `delivery_guarantee = at_least_once`, the sink writes with an idempotent producer without transactions, and a restart may write again messages that are already in Kafka.
- **Synchronous sink** (`TAtLeastOnceKafkaSink`): at-least-once — each epoch waits for Kafka to acknowledge its records. If the epoch does not commit after a successful write, the same records are written again on restart.

For more details, see [Kafka extension](https://ytsaurus.tech/docs/en/flow/extensions/kafka.md#guarantees).


### ClickHouse {#clickhouse-guarantees}

The ClickHouse extension provides three guarantee levels across four sink classes; your choice directly affects message processing guarantees:

- **Exactly-once sinks** (`TClickHouseBatchingSink`, recommended by default, and `TShardedClickHouseBatchingSink`): messages are stored in `output_messages` and deterministically grouped into batches by the same `MessageId` boundaries. In an unsharded sink, `insert_deduplication_token` equals the batch’s maximum `MessageId`. With `shard_hosts`, each shard receives that batch token with a `:<shard name>` suffix. A byte-identical replay carries the **same** deduplication token even after `group_by` repartitioning, so exactly-once survives repartitioning. ClickHouse deduplicates the repeat by the token.
- **At-least-once sink** (`TAtLeastOnceClickHouseSink`): the epoch batch is written synchronously during processing without intermediate storage and without a deduplication token. Lower latency and load on YTsaurus. A failed write prevents the epoch from committing, so messages are not lost. Known transient errors are retried without a limit; unclassified errors are attempted at most `max_insert_attempts` times in total (10 by default), and permanent errors fail immediately. On failure after a successful write but before epoch commit, the same batch is written again, so duplicates are possible.
- **At-most-once sink** (`TAtMostOnceClickHouseSink`): the guarantee applies in both strategy modes. With the default static setting `at_most_once_strategy.enabled = false`, pending messages stay in `output_messages` and use ordered delivery, so a connection or session failure before `INSERT` starts may be retried. Set the option to `true` to send each message independently through a bounded in-memory queue without waiting for delivery. Its dynamic limit is `at_most_once_strategy.total_queue_bytes_limit`; the sink logs a warning when it drops messages on overflow. In either mode, once `INSERT` has started, a failed insert is acknowledged instead of retried. Each message generates a separate `INSERT`, so to avoid creating many small parts in ClickHouse, aggregate the stream with batching upstream in the pipeline.

Exactly-once requires one of the ClickHouse table engines that deduplicate inserted blocks (see [Data Replication](https://clickhouse.com/docs/engines/table-engines/mergetree-family/replication) and [SharedMergeTree](https://clickhouse.com/docs/cloud/reference/shared-merge-tree) in the ClickHouse documentation):

- `ReplicatedMergeTree`
- `ReplicatedReplacingMergeTree`
- `ReplicatedSummingMergeTree`
- `ReplicatedAggregatingMergeTree`
- `SharedMergeTree`

Each sink instance creates its writer session lazily on the first write and keeps it until the instance stops. Complete target metadata validation runs when that session starts. Dynamic reconfiguration recreates clients when the write timeout changes and updates the relevant deduplication-window checks, but it does not repeat engine, schema, or replication-identity validation. After changing the target table, pause and start the pipeline so new sink instances validate it. `Distributed` tables, engines outside the `MergeTree` family, and multi-host or failover targets within a shard that cannot be proven to identify the same logical table are rejected. A single-host, non-replicated `MergeTree` without block deduplication is accepted with a warning, but exactly-once degrades to at-least-once. The deduplication window is finite and gets evicted: if a replay reaches ClickHouse later than the deduplication token’s eviction, exactly-once also degrades to at-least-once. If a known deduplication window is shorter than `replay_horizon`, the sink writes a structured warning with the attributes `Database`, `Table`, `DedupWindowSetting`, `ReplayHorizon`, and `DedupWindow`.

Summary table of guarantees by sink class:

#|
|| | **Exactly-once sink** | **At-least-once sink** | **At-most-once sink** ||
|| Intermediate storage in YTsaurus | Yes (`output_messages`) | No | Yes with `at_most_once_strategy.enabled = false`; no with `true` ||
|| Deduplication in ClickHouse | `insert_deduplication_token` (max `MessageId`; with `shard_hosts`, plus `:<shard name>`) | No | No ||
|| Latency | Higher (extra write) | Lower | Lower ||
|| Loss on failure | No | No | Possible after `INSERT` starts; with `enabled = true`, also on queue overflow ||
|| Duplicates on failure | No | Possible | No ||
|#

For more details, see [ClickHouse extension](https://ytsaurus.tech/docs/en/flow/extensions/clickhouse.md).

### Service log {#servicelog-guarantees}

- **Source**: exactly-once — deduplication by read ranges. The service log is an infinite source that cycles through the table. Exactly-once guarantees apply within each cycle.
- No sink. The sorted dynamic table used as a service log source is often populated by the pipeline itself via [external state](https://ytsaurus.tech/docs/en/flow/concepts/stateful.md#external-state) operations. For example, a `TTransformComputation` type computation updates table records at each epoch.

For more details, see [Service log](https://ytsaurus.tech/docs/en/flow/connectors/servicelog.md).


## Fault tolerance {#fault-tolerance}

- The [pipeline](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#pipeline) survives the failure of individual machines and data centers.
- While the pipeline remains active and scheduling continues, if a job fails, the controller removes it, aborts its [Lease](#lease) if it is still active, and schedules the necessary replacement jobs for the affected current partitions, with a new Lease for each replacement job. See [Lease transaction](#lease) for the fencing and recovery sequence.
- Internal streams are stored in YTsaurus dynamic tables — data loss isn’t possible during normal storage operation.
- [Automatic partition balancing](https://ytsaurus.tech/docs/en/flow/about.md) redistributes load when the cluster topology changes.

## See also

- [Flow overview](https://ytsaurus.tech/docs/en/flow/about.md)
- [Message processing order](https://ytsaurus.tech/docs/en/flow/concepts/ordering.md)
- [Timers](https://ytsaurus.tech/docs/en/flow/concepts/timers.md)
- [Stateful processing](https://ytsaurus.tech/docs/en/flow/concepts/stateful.md)
- [Connectors](https://ytsaurus.tech/docs/en/flow/connectors/about.md)
<!-- endsource: en/_includes/flow/concepts/guarantees.md -->