Processing guarantees in YTsaurus Flow

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 is updated atomically along with message processing.

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

Exactly-once semantics

How it works

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

1. Lease transaction

A Lease is a controller-owned master transaction that the controller creates for each 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 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. 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, provide output data atomicity or state atomicity, or cover arbitrary external side effects.

2. Input message deduplication

• 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.
• Sources (source): deduplication by offsets — consumer offsets advance only after a successful epoch commit.
• Timers: the result of timer processing is committed in the same transaction as the timer’s deletion.

3. Output data atomicity

The way output messages are delivered depends on the 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 pattern with deduplication in the target system, if it supports it.

What the developer must do

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 requirements depend on the computation type.

• For Swift 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.

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

Updates to Flow’s internal state (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

• Internal streams between computations — exactly-once.
• State updates — atomic with message processing.
• Timer processing and deletion — in the same transaction.
• Advancing source offsets — only after a successful epoch commit.

At-least-once semantics

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

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

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' MessageIds, 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 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

Flow doesn’t guarantee global message processing order. However, there are guarantees for derived messages with matching keys within a single lineage chain.

For details, see the Message processing order section.

How connectors affect guarantees

Each connector 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)

• 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.

Static Table

• 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.

HTTP

• 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.

Kafka

• 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.

ClickHouse

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 and SharedMergeTree 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:

For more details, see ClickHouse extension.

Service log

• 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 operations. For example, a TTransformComputation type computation updates table records at each epoch.

For more details, see Service log.

Fault tolerance

• The 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 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 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 redistributes load when the cluster topology changes.

See also

• Flow overview
• Message processing order
• Timers
• Stateful processing
• Connectors