> **Can't find what you're looking for?** Use `search_docs` on the docs MCP server at `https://docs.ampdb.io/api/mcp` to find what you need. # Streaming Reference Status: Unstable. The streaming classes and types are in the `amp.streaming` package. You configure [parallel streaming](/sdks/python/parallel-streaming) with `ParallelConfig`, and get the others back from streaming loads, like a reorg's `BlockRange`s. The rest are what `load()` and [loaders](/sdks/python/reference/loaders#dataloader) build on, which you can use yourself: to run a parallel load without `load()`, or to test a loader with batches you make. [Streaming](/sdks/python/streaming) explains how they fit together. ```python from amp.streaming import ( BatchIdentifier, BatchMetadata, BlockRange, BlockRangePartitionStrategy, InMemoryStreamStateStore, LMDBStreamStateStore, NullStreamStateStore, ParallelConfig, ParallelStreamExecutor, ProcessedBatch, QueryPartition, ReorgAwareStream, ResponseBatch, ResumeWatermark, StreamingResultIterator, StreamStateStore, ) from amp.streaming.resilience import BackPressureConfig, RetryConfig ``` ## `ParallelConfig` ```python ParallelConfig( num_workers: int, table_name: str, min_block: int = 0, max_block: int | None = None, partition_size: int | None = None, block_column: str = 'block_num', stop_on_error: bool = False, reorg_buffer: int = 200, retry_config: RetryConfig | None = None, back_pressure_config: BackPressureConfig | None = None, ) ``` How a [parallel load](/sdks/python/parallel-streaming) splits and loads its block range. Pass it to `load()` as `parallel_config`, with `stream=True`. | Field | Type | Default | Description | | ---------------------- | ----------------------------------------------------- | ----------- | ------------------------------------------------------------------------------------------------- | | `num_workers` | `int`, 1 or more | Required | How many partitions load at once. | | `table_name` | `str` | Required | The table to partition, as SQL names it, like `'"acme/eth_mainnet".transactions'`. | | `min_block` | `int` | `0` | The first block to load. | | `max_block` | `int` or `None` | `None` | The block to stop before. `None` loads up to the last block with data, then streams: hybrid mode. | | `partition_size` | `int`, 1 or more, or `None` | `None` | How many blocks each partition has. `None` makes one partition per worker. | | `block_column` | `str` | `block_num` | The column with the block number. | | `stop_on_error` | `bool` | `False` | Stops every worker, and raises the error, when a partition fails. | | `reorg_buffer` | `int` | `200` | In hybrid mode, how many blocks before the last one loaded the stream starts. | | `retry_config` | [`RetryConfig`](/sdks/python/reference/streaming#retryconfig) or `None` | `None` | Retries for every worker's loader. `None` uses the loader's own. | | `back_pressure_config` | [`BackPressureConfig`](/sdks/python/reference/streaming#backpressureconfig) or `None` | `None` | Back pressure for every worker's loader. `None` uses the loader's own. | Raises `ValueError` when `num_workers` or `partition_size` is less than 1 (`num_workers must be >= 1, got 0`), when `max_block` isn't greater than `min_block` (`min_block (10) must be < max_block (5)`), or when `table_name` is empty (`table_name is required`). `get_resilience_config()` returns `retry_config` and `back_pressure_config` as the `resilience` section of a loader's configuration, like `{'resilience': {'retry': {'enabled': True, 'max_retries': 5, …}}}`, leaving out the ones that are `None`. It returns `{}` when both are. ## `ParallelStreamExecutor` ```python ParallelStreamExecutor(client: Client, config: ParallelConfig) ``` Runs a [parallel load](/sdks/python/parallel-streaming): what `load(stream=True, parallel_config=config)` creates and runs for you. Create one yourself to run a parallel load from a query string, without `client.sql()`. It starts a pool of `config.num_workers` threads, which it shuts down when the load ends, so each executor runs one load. ### `execute_parallel_stream()` ```python execute_parallel_stream( user_query: str, destination: str, connection_name: str, load_config: dict | None = None, ) -> Iterator[LoadResult] ``` Splits the configured block range into [partitions](/sdks/python/reference/streaming#querypartition), creates the destination from a sample of the query's results, then loads the partitions, as [How it works](/sdks/python/parallel-streaming#how-it-works) describes. Without a `max_block`, it then streams, in [hybrid mode](/sdks/python/parallel-streaming#hybrid-mode-catch-up-then-stream). | Parameter | Type | Default | Description | | ----------------- | ---------------- | -------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `user_query` | `str` | Required | The SQL query. Each partition adds its block filter to it (see [Write a query that partitions](/sdks/python/parallel-streaming#write-a-query-that-partitions)). | | `destination` | `str` | Required | What the loader writes to, like `load()`'s `destination`. | | `connection_name` | `str` | Required | A configured connection's name. Inline configurations aren't supported. | | `load_config` | `dict` or `None` | `None` | [Load options](/sdks/python/loaders#load-options), like `{'mode': 'append'}`, and the loader's own. The executor adds to the dictionary you pass. | Returns an iterator of [`LoadResult`](/sdks/python/reference/loaders#loadresult)s: one per partition, in the order they finish, then, in hybrid mode, one per batch of the stream. Nothing runs until you loop over it. A partition's result has its `partition_id`, `batches_processed`, `worker_thread`, and its [`QueryPartition`](/sdks/python/reference/streaming#querypartition)'s `metadata` as `partition_metadata`. A partition that fails gives a failed result, with its `partition_id` and `partition_metadata`, and the others carry on. With `stop_on_error=True`, it raises the partition's error instead. When it can't find the last block, in hybrid mode, or can't split the range, it gives one failed result, with the error `Failed to detect max block: …` or `Partition creation failed: …`. A second call on the same executor raises `RuntimeError: cannot schedule new futures after shutdown`. This script runs the [Parallel Streaming](/sdks/python/parallel-streaming#load-a-block-range-in-parallel) example's load with an executor, into the `local_postgres` connection it configures, and prints each partition's ID, rows, and batches: ```python [parallel_executor.py] from amp import Client from amp.streaming import ParallelConfig, ParallelStreamExecutor client = Client('grpc://localhost:1602') config = ParallelConfig( num_workers=4, table_name='"acme/eth_mainnet".transactions', min_block=24000000, max_block=24000100, ) executor = ParallelStreamExecutor(client, config) results = executor.execute_parallel_stream( 'SELECT block_num, tx_index, tx_hash, gas_used FROM "acme/eth_mainnet".transactions', destination='transactions', connection_name='local_postgres', ) for result in results: print(result.metadata['partition_id'], result.rows_loaded, result.metadata['batches_processed']) ``` :::terminal ```bash python parallel_executor.py 2> /dev/null ``` ``` 3 3614 2 2 4245 2 1 4665 2 0 4629 2 ``` ::: `2> /dev/null` hides the workers' progress lines, which they write to standard error. The partitions finish in any order. ## `QueryPartition` ```python QueryPartition(partition_id: int, start_block: int, end_block: int, block_column: str = 'block_num') ``` One part of a parallel load's block range, which one worker loads. | Member | Type | Description | | -------------- | ------ | ------------------------------------------------------------------------------------------------------------------ | | `partition_id` | `int` | Its number, from 0. | | `start_block` | `int` | The first block. | | `end_block` | `int` | The block it stops before. | | `block_column` | `str` | The column its filter uses. | | `metadata` | `dict` | `start_block`, `end_block`, `block_column`, and `partition_size`, its number of blocks. Partition results have it. | ## `BlockRangePartitionStrategy` ```python BlockRangePartitionStrategy(table_name: str, block_column: str = 'block_num') ``` How a parallel load splits its range into [`QueryPartition`](/sdks/python/reference/streaming#querypartition)s, and filters the query for each. Use it to see the partitions and queries a [`ParallelConfig`](/sdks/python/reference/streaming#parallelconfig) makes, before you load. | Method | Returns | Description | | --------------------------------------------- | ------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `create_partitions(config)` | list of `QueryPartition` | Splits `config`'s range, from `min_block` up to `max_block`, into one partition per worker, or into partitions of `partition_size` blocks. Uses `config.block_column`. Needs a `max_block`: raises `TypeError` without one. | | `wrap_query_with_partition(query, partition)` | `str` | The query with the partition's block filter: added to its `WHERE` clause, in parentheses, or as a new `WHERE` clause, before any `GROUP BY`, `ORDER BY`, `LIMIT`, or `SETTINGS`. | ```python [partitions.py] from amp.streaming import BlockRangePartitionStrategy, ParallelConfig config = ParallelConfig( num_workers=4, table_name='"acme/eth_mainnet".blocks', min_block=24000000, max_block=24000100, partition_size=30, ) strategy = BlockRangePartitionStrategy(config.table_name) partitions = strategy.create_partitions(config) for partition in partitions: print(partition.partition_id, partition.metadata) query = 'SELECT block_num, gas_used FROM "acme/eth_mainnet".blocks WHERE gas_used > 0 ORDER BY block_num' print(strategy.wrap_query_with_partition(query, partitions[-1])) ``` :::terminal ```bash python partitions.py ``` ``` 0 {'start_block': 24000000, 'end_block': 24000030, 'block_column': 'block_num', 'partition_size': 30} 1 {'start_block': 24000030, 'end_block': 24000060, 'block_column': 'block_num', 'partition_size': 30} 2 {'start_block': 24000060, 'end_block': 24000090, 'block_column': 'block_num', 'partition_size': 30} 3 {'start_block': 24000090, 'end_block': 24000100, 'block_column': 'block_num', 'partition_size': 10} SELECT block_num, gas_used FROM "acme/eth_mainnet".blocks WHERE gas_used > 0 AND (block_num >= 24000090 AND block_num < 24000100) ORDER BY block_num ``` ::: It needs no Amp: it only computes the partitions and queries. ## `RetryConfig` ```python from amp.streaming.resilience import RetryConfig RetryConfig( enabled: bool = True, max_retries: int = 5, initial_backoff_ms: int = 2000, max_backoff_ms: int = 120000, backoff_multiplier: float = 2.0, jitter: bool = True, ) ``` How a loader retries a batch that failed with an error that may pass. A loader's configuration sets the same fields under `resilience.retry`: see [Retries](/sdks/python/loaders#retries), which describes each. ## `BackPressureConfig` ```python from amp.streaming.resilience import BackPressureConfig BackPressureConfig( enabled: bool = True, initial_delay_ms: int = 0, max_delay_ms: int = 5000, adapt_on_429: bool = True, adapt_on_timeout: bool = True, recovery_factor: float = 0.9, ) ``` How a loader slows down when the destination limits it. A loader's configuration sets the same fields under `resilience.back_pressure`: see [Retries](/sdks/python/loaders#retries). ## `BlockRange` ```python BlockRange(network: str, start: int, end: int, hash: str | None = None, prev_hash: str | None = None) ``` A range of blocks on one network, both ends included. A reorg's `LoadResult` lists the ranges whose rows the loader deleted in `invalidation_ranges`. | Field | Type | Description | | ----------- | --------------- | ---------------------------------------------------------------- | | `network` | `str` | The network, like `'mainnet'`. | | `start` | `int` | The first block. | | `end` | `int` | The last block. Raises `ValueError` when it's less than `start`. | | `hash` | `str` or `None` | The hash of the last block, from a streaming query. | | `prev_hash` | `str` or `None` | The hash of the block before the first, from a streaming query. | | Method | Returns | Description | | ------------------------- | ------------ | ------------------------------------------------------------------------------------ | | `overlaps_with(other)` | `bool` | Whether the two ranges share a block, on the same network. | | `invalidates(other)` | `bool` | The same as `overlaps_with(other)`. | | `contains_block(number)` | `bool` | Whether a block is in the range. | | `merge_with(other)` | `BlockRange` | The range from the first start to the last end. Raises `ValueError` across networks. | | `to_dict()` | `dict` | `{'network': …, 'start': …, 'end': …}`, with `hash` and `prev_hash` when set. | | `BlockRange.from_dict(d)` | `BlockRange` | The reverse of `to_dict()`. Also reads Amp's format, with `numbers: {start, end}`. | ## `ResumeWatermark` ```python ResumeWatermark(ranges: list[BlockRange], timestamp: str | None = None, sequence: int | None = None) ``` The block ranges a stream loaded last, one per network, with their hashes. Passed to `load()` as `resume_watermark`, it lets reorg detection compare the stream's first batches with them. Amp still streams from the start of the query's results (see [Known issues](/sdks/python/streaming#restarting-loads-everything-again)). `to_json()` returns it as a JSON string, and `ResumeWatermark.from_json(text)` reads one back, to save it between runs. ## `ResponseBatch` ```python ResponseBatch( data: pyarrow.RecordBatch, metadata: BatchMetadata, is_reorg: bool = False, invalidation_ranges: list[BlockRange] | None = None, ) ``` A batch of a stream, as loaders receive it. | Field | Type | Description | | --------------------- | --------------------------------- | -------------------------------------------- | | `data` | `pyarrow.RecordBatch` | Its rows. Empty for a reorg. | | `metadata` | [`BatchMetadata`](/sdks/python/reference/streaming#batchmetadata) | The block ranges it covers. | | `is_reorg` | `bool` | Whether it reports a reorg rather than rows. | | `invalidation_ranges` | list of `BlockRange` or `None` | For a reorg, the ranges to undo. | | `num_rows` | `int` | How many rows `data` has. | | `networks` | list of `str` | The networks its ranges cover. | Make one with these instead of the constructor, to give a loader a stream you build, like the [Custom Loaders](/sdks/python/loaders/custom#reorgs) page's test does: | Method | Returns | Description | | ------------------------------------------------ | --------------- | ------------------------------------------------------------------------------------------ | | `ResponseBatch.data_batch(data, metadata)` | `ResponseBatch` | A batch of rows, from a `pyarrow.RecordBatch` and its [`BatchMetadata`](/sdks/python/reference/streaming#batchmetadata). | | `ResponseBatch.reorg_batch(invalidation_ranges)` | `ResponseBatch` | A reorg of the given `BlockRange`s, with no columns, no rows, and no ranges in `metadata`. | ```python import pyarrow as pa from amp.streaming import BatchMetadata, BlockRange, ResponseBatch rows = pa.record_batch({'block_num': pa.array([100, 101], type=pa.uint64())}) batch = ResponseBatch.data_batch( rows, BatchMetadata([BlockRange('mainnet', 100, 101, hash='0xb', prev_hash='0xa')], ranges_complete=True) ) reorg = ResponseBatch.reorg_batch([BlockRange('mainnet', 100, 101)]) ``` ## `BatchMetadata` ```python BatchMetadata(ranges: list[BlockRange], ranges_complete: bool = False, extra: dict | None = None) ``` The block ranges a batch of a stream covers, as Amp sends them. | Field | Type | Description | | ----------------- | -------------------- | --------------------------------------------------------------------------------- | | `ranges` | list of `BlockRange` | One per network, with its hashes. | | `ranges_complete` | `bool` | Whether this batch completes its ranges: the SDK records them as loaded after it. | | `extra` | `dict` or `None` | Anything else Amp sent. | `BatchMetadata.from_flight_data(metadata_bytes)` reads the JSON that Amp sends with each batch of a streaming query, as `bytes` or a `pyarrow.Buffer`, and each of its ranges with [`BlockRange.from_dict()`](/sdks/python/reference/streaming#blockrange). It returns no ranges, with the error in `extra['parse_error']`, when the text isn't JSON or a range lacks a key: `network`, or `start` or `end` in `to_dict()`'s format. Other malformed metadata raises instead, for example: | Metadata | Raises | | -------------------------------------------- | -------------------- | | Bytes that aren't UTF-8 | `UnicodeDecodeError` | | JSON that isn't an object, like `[]` | `AttributeError` | | A range that isn't an object | `TypeError` | | A range's `numbers` without `start` or `end` | `TypeError` | | A range whose `start` is after its `end` | `ValueError` | [`StreamingResultIterator`](/sdks/python/reference/streaming#streamingresultiterator) uses it, and catches these errors: it logs a warning and gives the batch no ranges. ```python [batch_metadata.py] from amp.streaming import BatchMetadata metadata = BatchMetadata.from_flight_data( b'{"ranges":[{"numbers":{"start":24000000,"end":24000099},"network":"mainnet",' b'"hash":"0x6ed4b3511195052eea47925fe5035f5e3fc528440eed62649ba82058a1b70e40",' b'"prev_hash":"0xe8f20171e085c248ac4d60b65ddeb9c1bd679149cc1be9925ad58799dc380b25",' b'"timestamp":1765585559}],"ranges_complete":true}' ) print(metadata.ranges[0].start, metadata.ranges[0].end, metadata.ranges_complete) print(BatchMetadata.from_flight_data(b'not JSON')) try: BatchMetadata.from_flight_data(b'{"ranges":[{"network":"mainnet","numbers":{"start":100}}]}') except TypeError as e: print(f'TypeError: {e}') ``` :::terminal ```bash python batch_metadata.py ``` ``` 24000000 24000099 True BatchMetadata(ranges=[], ranges_complete=False, extra={'parse_error': 'Expecting value: line 1 column 1 (char 0)'}) TypeError: '>' not supported between instances of 'int' and 'NoneType' ``` ::: That's the metadata of the Quickstart's 100 blocks, as a stream of `"acme/eth_mainnet".blocks` sends it. ## `StreamingResultIterator` ```python StreamingResultIterator(flight_reader: pyarrow.flight.FlightStreamReader) ``` Turns a streaming query's Arrow Flight results into [`ResponseBatch`](/sdks/python/reference/streaming#responsebatch)es, reading each batch's block ranges with [`BatchMetadata.from_flight_data()`](/sdks/python/reference/streaming#batchmetadata). It's an iterator: loop over it, or call `next()`, to get each batch. `load(stream=True)` makes one from the query's Flight stream for you. It ends at Amp's first keep-alive message, 30 seconds after the stream starts, which pyarrow takes for the end of the stream (see [Streams end after 30 seconds](/sdks/python/streaming#streams-end-after-30-seconds)). In your program's main thread, it handles Ctrl+C by ending the stream, without an error. The handler stays in place after the stream ends, so Ctrl+C then does nothing (see [Ctrl+C ends a stream without an error](/sdks/python/streaming#ctrlc-ends-a-stream-without-an-error)). `close()` ends the stream too, and so does leaving a `with` block: `with StreamingResultIterator(reader) as stream:`. ## `ReorgAwareStream` ```python ReorgAwareStream(stream_iterator: StreamingResultIterator, resume_watermark: ResumeWatermark | None = None) ``` Wraps a `StreamingResultIterator`, and yields a reorg `ResponseBatch` instead of a batch whose ranges reveal a reorg, as [Reorgs](/sdks/python/streaming#reorgs) describes. With a `resume_watermark`, it starts from its ranges. `close()`, or leaving a `with` block, ends the stream. A batch reveals a reorg when, on the same network as the batch before it, its range is different and starts at or before the previous one's end, or its `prev_hash` isn't the previous range's `hash`. The reorg's range goes from the previous range's start to the later of the two ends. The batch that revealed it isn't yielded: `load()` starts the stream again, which sends it again. It wraps any iterator of `ResponseBatch`es, so you can see what it reports for ranges you make: ```python [reorg_stream.py] import pyarrow as pa from amp.streaming import BatchMetadata, BlockRange, ReorgAwareStream, ResponseBatch def batch(start, end, block_hash, prev_hash): data = pa.record_batch({'block_num': pa.array(range(start, end + 1), type=pa.uint64())}) block_range = BlockRange('mainnet', start, end, hash=block_hash, prev_hash=prev_hash) return ResponseBatch.data_batch(data, BatchMetadata([block_range], ranges_complete=True)) batches = [ batch(100, 101, '0xb', '0xa'), batch(102, 103, '0xd', '0xb'), batch(102, 103, '0xf', '0xe'), # the same blocks again, with new hashes ] for response in ReorgAwareStream(iter(batches)): if response.is_reorg: print('Reorg:', response.invalidation_ranges) else: print('Rows:', response.data.column('block_num').to_pylist()) ``` :::terminal ```bash python reorg_stream.py ``` ``` Rows: [100, 101] Rows: [102, 103] Reorg: [BlockRange(network='mainnet', start=102, end=103, hash='0xd', prev_hash=None)] ``` ::: ## Stream state stores Where a loader records the batches it loaded, as its `state_store`. Each identifies a batch by a [`BatchIdentifier`](/sdks/python/reference/streaming#batchidentifier), and keeps its records per connection name, destination, and network. | Class | Description | | -------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `StreamStateStore` | The base class, with the methods below. | | `InMemoryStreamStateStore` | Keeps them in memory, until your program exits. Every loader's default. | | `NullStreamStateStore` | Keeps nothing. Loaders use it with `state.enabled` set to `False`. | | `LMDBStreamStateStore` | Keeps them in an LMDB database on disk: `LMDBStreamStateStore(connection_name, data_dir='.amp_state', map_size=10 * 1024**3, sync=True)`. `close()`, or leaving a `with` block, closes the database. Needs the `lmdb` package: without it, `LMDBStreamStateStore` is `None`. No loader uses it by default. | | Method | Returns | Description | | -------------------------------------------------------------------------- | --------------------------- | -------------------------------------------------------------------------------------------------------- | | `is_processed(connection_name, table_name, batch_ids)` | `bool` | Whether all the batches are recorded. | | `mark_processed(connection_name, table_name, batch_ids)` | `None` | Records the batches. | | `get_resume_position(connection_name, table_name, detect_gaps=False)` | `ResumeWatermark` or `None` | The last block recorded on each network. With `detect_gaps=True`, also the ranges missing before it. | | `invalidate_from_block(connection_name, table_name, network, from_block)` | list of `BatchIdentifier` | Removes the batches that end at or after a block, and returns them, for the loader to delete their rows. | | `cleanup_before_block(connection_name, table_name, network, before_block)` | `None` | Removes the records of batches that end before a block, to keep the state small. | ## `BatchIdentifier` ```python BatchIdentifier(network: str, start_block: int, end_block: int, end_hash: str, start_parent_hash: str = '') ``` A batch's identity: its network, block range, and hashes. `BatchIdentifier.from_block_range(block_range)` makes one from a `BlockRange`; for a range without hashes, like a [parallel](/sdks/python/parallel-streaming) partition's, it uses a hash of the range instead. | Member | Type | Description | | -------------------------- | ----------------- | ------------------------------------------------------------------------------------------------------------- | | `unique_id` | `str` | 16 hexadecimal characters, from a hash of all its fields: the `_amp_batch_id` that loaders write to each row. | | `position_key` | `(str, int, int)` | `(network, start_block, end_block)`. | | `to_block_range()` | `BlockRange` | The batch's range. | | `overlaps_or_after(block)` | `bool` | Whether the batch ends at or after a block: whether a reorg from that block undoes it. | ## `ProcessedBatch` ```python ProcessedBatch(batch_id: BatchIdentifier, processed_at: datetime = now, reorg_invalidation: bool = False) ``` A record of a loaded batch, for a [stream state store](/sdks/python/reference/streaming#stream-state-stores) you write that keeps its records in a database. The SDK's own stores don't use it. | Member | Type | Description | | ----------------------------- | ------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------ | | `batch_id` | [`BatchIdentifier`](/sdks/python/reference/streaming#batchidentifier) | The batch. | | `processed_at` | `datetime` | When it was loaded. Defaults to the current time, in UTC. | | `reorg_invalidation` | `bool` | Whether a reorg undid it. | | `to_dict()` | `dict` | The batch's fields, its `unique_id`, `processed_at` as an ISO 8601 string, and `reorg_invalidation`, to store as a row or a JSON document. | | `ProcessedBatch.from_dict(d)` | `ProcessedBatch` | The reverse of `to_dict()`. | ```python [processed_batch.py] from amp.streaming import BatchIdentifier, BlockRange, ProcessedBatch block_range = BlockRange( 'mainnet', 24000000, 24000099, hash='0x6ed4b3511195052eea47925fe5035f5e3fc528440eed62649ba82058a1b70e40' ) record = ProcessedBatch(BatchIdentifier.from_block_range(block_range)) saved = record.to_dict() print(saved) print(ProcessedBatch.from_dict(saved) == record) ``` :::terminal ```bash python processed_batch.py ``` ``` {'network': 'mainnet', 'start_block': 24000000, 'end_block': 24000099, 'end_hash': '0x6ed4b3511195052eea47925fe5035f5e3fc528440eed62649ba82058a1b70e40', 'start_parent_hash': '', 'unique_id': 'aa8b21f1b2f1a347', 'processed_at': '2026-10-30T21:55:09.191577+00:00', 'reorg_invalidation': False} True ``` ::: Your `processed_at` differs. The `unique_id` doesn't: it depends only on the batch's fields.