mirror of
https://github.com/temporalio/temporal.git
synced 2026-08-30 18:41:49 -07:00
3e923ba44c
> Part 1 of a planned 5-PR series building toward replication stream
namespace isolation (a restructuring of #10147): read buffer → reader
group → lane protocol → isolation manager → sender isolation. This PR
stands on its own; follow-ups will be opened as each lands review.
**Next in series: #11302** (reader group).
## What changed?
A shard-scoped read-through buffer over the tip of the replication task
queue, sitting inside the ack manager's `GetReplicationTasksIter`. Every
replication stream sender on a shard — one per remote cluster, with one
iterator per priority lane — scans the same queue, so each new task page
was previously read from persistence once per scanner. With the buffer,
overlapping tip scans share one persistence read; only readers below the
buffered range (deep catch-up, lanes that have lagged out of coverage)
fall through to persistence.
Coverage is a contiguous task-id interval established by persistence
pages: within it the buffer is authoritative, so absence of a task means
the range holds none. An empty persistence page with a non-empty
continuation token (legal, e.g. under Cassandra paging) is NOT
authoritative — the fetch keeps paging until tasks arrive or the token
runs out. Rows are immutable and the queue is append-only, so there is
no invalidation; eviction just shrinks coverage from the front. Reads
are bounded by the shard's exclusive-high read watermark, so covered
ranges are stable once established.
**Ownership:** the buffer stores SERIALIZED rows and deserializes per
serve, so every reader receives fresh task structs it exclusively owns —
exactly what a persistence read would have produced. This matters
because downstream converters mutate tasks in place (e.g.
`SyncVersionedTransitionTask` equivalents get IDs assigned via
`AddTasks`); handing multiple senders pointers to shared structs would
be a data race. The serve-time deserialization replaces the persistence
read the reader would otherwise have done, so it is not added cost
relative to the unbuffered path. Serialize/deserialize failures are
never silent: both are error-logged (they indicate a bug — e.g. a task
type missing serializer support — or broken persistence data); a
serialize failure serves the page uncached, a deserialize failure drops
the buffer's coverage entirely and falls back to persistence.
Capacity is `ReplicationStreamReadBufferSize` tasks per shard (default 0
= disabled); disabling at runtime releases the buffered rows. The buffer
holds slim queue rows (task metadata) for every priority — event
payloads only enter the pipeline at send-time conversion — so memory
cost is a few hundred bytes per row.
Observability (to drive future sizing/sharing decisions):
`replication_stream_read_buffer_hits` / `_misses` count pages served
from memory vs. fetched from persistence while the buffer is enabled
(misses are counted only after a successful fetch), and
`replication_stream_read_buffer_miss_lag` records, for misses below the
buffered range, how far below coverage the read began (in task ids).
Small lag values mean a larger buffer would convert those misses to
hits; large values mean readers deep in backlog, where no tip buffer
helps.
## Why?
A standalone win for the code as it is today, with no dependency on the
rest of this series: any multi-cluster mesh already pays `remote
clusters × priority lanes` read amplification on the queue tip, and the
buffer collapses those overlapping scans into one persistence read per
page. It additionally unlocks the later PRs in the series: per-namespace
isolation lanes multiply the number of concurrent scanners, and with the
buffer their tip scans become in-memory filter passes instead of extra
persistence load.
## How did you test it?
- [x] built
- [x] added new unit test(s) — pass-through when disabled, memory
serving for second readers and partial overlaps, contiguous coverage
extension, below-coverage fall-through without cache disturbance, gap
restart at a newer tip, front eviction, truncated-page authority bounds,
hit/miss/lag metric emission, per-reader ownership of served rows,
runtime disable releasing state, empty-page-with-token continuation in
`GetReplicationTasksIter`, and a `-race` concurrent-readers stress test
over a moving tip
- [x] covered by existing tests — the xdc isolation test later in the
series runs with the buffer enabled
## Potential risks
Default-off. The main correctness surface is coverage bookkeeping
(serving a range the buffer isn't authoritative for); the
coverage-interval design plus the truncated-fetch and empty-page-token
tests target exactly that. Serve-time deserialization guarantees no
cross-reader object sharing, and round-trip failures are loud (error
logs) rather than silently degrading. Memory is strictly bounded by the
row-count cap and released on disable.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
<!-- CURSOR_SUMMARY -->
---
> [!NOTE]
> **Medium Risk**
> Changes replication task loading and coverage bookkeeping in the ack
manager; default-off but incorrect coverage could skip or mis-serve
tasks when the buffer is enabled.
>
> **Overview**
> Adds a **shard-scoped read-through buffer** on the replication task
queue tip so overlapping scans from every remote cluster and priority
lane can share one persistence read instead of amplifying reads per
scanner.
>
> **`readBuffer`** (`read_buffer.go`) tracks contiguous task-id
coverage, stores serialized slim queue rows, and deserializes per serve
so each reader gets owned task structs (downstream code mutates tasks in
place). Capacity is **`ReplicationStreamReadBufferSize`** (default **0**
= disabled); disabling at runtime clears buffered state.
>
> **`GetReplicationTasksIter`** in the ack manager routes reads through
the buffer and tightens persistence paging: empty pages with a
continuation token keep paging until tasks arrive or the token is empty;
truncated pages only extend coverage through the last returned task id.
>
> New metrics: **`replication_stream_read_buffer_hits`**, **`_misses`**,
and **`_miss_lag`** for tuning buffer size and observing catch-up
behavior.
>
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
5836b28b30. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->
---------
Co-authored-by: Claude <noreply@anthropic.com>