ruststream-kinesis 0.7.0

//! [`KinesisSubscriber`]: shard discovery, leasing, and per-shard readers.
//!
//! The substantial work of this crate: a coordinator task lists the stream's shards, gates
//! children on their parents being fully consumed, takes leases through the store, and runs
//! one reader task per owned shard; readers poll `GetRecords`, feed the shared delivery
//! channel, renew their lease (stopping immediately when fenced), and close their shard at
//! `SHARD_END` once every delivery has settled.

use std::collections::HashMap;
use std::mem;
use std::num::NonZeroUsize;
use std::sync::Arc;
use std::time::Duration;

use std::sync::atomic::{AtomicI32, AtomicU64, Ordering};
use std::sync::{Mutex, MutexGuard};

use aws_sdk_kinesis::operation::get_records::GetRecordsError;
use aws_sdk_kinesis::operation::get_shard_iterator::builders::GetShardIteratorFluentBuilder;
use aws_sdk_kinesis::primitives::DateTime;
use aws_sdk_kinesis::types::{Shard, ShardIteratorType};
use futures::Stream;
#[cfg(feature = "testing")]
use futures::future::Either;
use ruststream::{BatchSubscriber, BufferedSubscriber, Subscriber};
use tokio::runtime::Handle;
use tokio::sync::{mpsc, oneshot};

use crate::broker::Core;
use crate::error::{KinesisError, sdk_err};
#[cfg(feature = "testing")]
use crate::in_process::{LogDeliveries, LogSeeker};
use crate::lease::{LeaseError, LeaseKey, LeaseStore, SHARD_END};
use crate::message::{KPL_MAGIC, KinesisMessage, KinesisPosition, Settle, Settlement};
use crate::stream::KinesisStream;
use crate::track::Watermark;

/// How often the coordinator re-lists shards (splits and merges change the set over time).
const SHARD_SYNC: Duration = Duration::from_secs(10);
/// How long a lease is valid without renewal, and the renewal cadence derived from it.
const LEASE_TTL: Duration = Duration::from_secs(10);
const RENEW_EVERY: Duration = Duration::from_secs(3);
/// How many deliveries may sit between the readers and the consumer.
const CHANNEL_CAPACITY: usize = 64;
/// Records one `GetRecords` call asks for while the subscription delivers single messages. A
/// batch subscription replaces it with the size its mount site named.
const DEFAULT_READ_LIMIT: i32 = 1000;
/// The service's own cap on one read.
const MAX_READ_LIMIT: i32 = 10_000;
/// How long a partial batch waits for more records before it goes out. Chosen once here because
/// the size is the mount site's and the deadline is the broker's; it is short next to the
/// second the reader waits between reads of an idle shard.
const BATCH_FILL_WINDOW: Duration = Duration::from_millis(50);

/// A position every shard of the subscription can open at: the stream-wide half of
/// [`KinesisPosition`]. Kept apart from the shard-scoped forms so that "install this for the
/// whole subscription" cannot be handed a position that only one shard understands.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum StreamStart {
    Horizon,
    Latest,
    Timestamp(u64),
}

/// The cursor one shard's reader opens with, in the shapes the service's iterator types take.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum ShardStart {
    /// A position shared by the whole subscription.
    Stream(StreamStart),
    /// Redelivers exactly this record: a captured, pinned position.
    At(String),
    /// Resumes after this record: a stored checkpoint.
    After(String),
}

/// A repositioning request delivered to one shard's reader.
pub(crate) struct ShardSeek {
    pub(crate) start: ShardStart,
    pub(crate) done: oneshot::Sender<Result<(), KinesisError>>,
}

/// One shard's live seek surface: the delivery-generation gate (bumped at enqueue, so
/// in-flight batches stamp stale) and the reader's command channel.
#[derive(Clone)]
pub(crate) struct ShardHandle {
    pub(crate) gate: Arc<AtomicU64>,
    pub(crate) tx: mpsc::UnboundedSender<ShardSeek>,
}

/// The subscription's seek surface, shared by the seeker, the coordinator, and every reader.
pub(crate) struct SeekState {
    /// The connection the subscription reads over. A seeker outlives its broker's `shutdown`,
    /// and a seek through it then must fail rather than reposition readers on a closed
    /// connection.
    core: Arc<Core>,
    /// The readers that can be repositioned right now, by shard id.
    shards: Mutex<HashMap<String, ShardHandle>>,
    /// The stream-wide position a seek installed, if any. Readers consult it when they fetch
    /// their first iterator, which is what makes a stream-wide seek reach shards that have no
    /// reader yet: the subscription opened microseconds ago (the `start_at(..)` case), or the
    /// shard is a child that only appears after a split.
    start: Mutex<Option<StreamStart>>,
}

pub(crate) type SeekBus = Arc<SeekState>;

impl SeekState {
    fn new(core: Arc<Core>) -> Self {
        Self {
            core,
            shards: Mutex::default(),
            start: Mutex::default(),
        }
    }

    fn register(&self, shard: String, handle: ShardHandle) {
        self.lock_shards().insert(shard, handle);
    }

    fn deregister(&self, shard: &str) {
        self.lock_shards().remove(shard);
    }

    fn handle(&self, shard: &str) -> Option<ShardHandle> {
        self.lock_shards().get(shard).cloned()
    }

    fn live(&self) -> Vec<ShardHandle> {
        self.lock_shards().values().cloned().collect()
    }

    fn install(&self, start: StreamStart) {
        *self.start.lock().expect("seek state mutex poisoned") = Some(start);
    }

    pub(crate) fn installed(&self) -> Option<StreamStart> {
        *self.start.lock().expect("seek state mutex poisoned")
    }

    fn lock_shards(&self) -> MutexGuard<'_, HashMap<String, ShardHandle>> {
        self.shards.lock().expect("seek state mutex poisoned")
    }
}

/// One channel item: the delivery (or error) plus its generation stamp. A seek bumps the
/// shard's gate, and items stamped under an older generation are discarded on the way out.
pub(crate) struct Stamped {
    stamp: Option<(u64, Arc<AtomicU64>)>,
    item: Result<KinesisMessage, KinesisError>,
}

impl Stamped {
    fn live(epoch: u64, gate: &Arc<AtomicU64>, item: Result<KinesisMessage, KinesisError>) -> Self {
        Self {
            stamp: Some((epoch, Arc::clone(gate))),
            item,
        }
    }

    fn unstamped(item: Result<KinesisMessage, KinesisError>) -> Self {
        Self { stamp: None, item }
    }

    fn current(&self) -> bool {
        self.stamp
            .as_ref()
            .is_none_or(|(epoch, gate)| *epoch == gate.load(Ordering::Acquire))
    }
}

/// A subscription to one Kinesis stream; yields [`KinesisMessage`]s from every owned shard.
///
/// Deliveries arrive one record at a time, because acknowledgement is a per-record checkpoint
/// against the shard's watermark, and a batch mount groups them: the size the mount site named
/// becomes the `GetRecords` limit every reader asks with, so a read never fetches more than one
/// batch's worth, and the batch itself is closed by that size or, when the shards had less to
/// give, 50 ms after its first record.
///
/// Dropping the subscriber stops the coordinator and every reader; unsettled records
/// redeliver from the last checkpoint when the leases are next taken.
pub struct KinesisSubscriber {
    stream: String,
    /// The live `GetRecords` limit, shared with every reader. This is how the mount site's batch
    /// size reaches the wire: [`BatchSubscriber::batches`] stores it before the first batch.
    limit: Arc<AtomicI32>,
    deliveries: BufferedSubscriber<Deliveries>,
}

impl std::fmt::Debug for KinesisSubscriber {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("KinesisSubscriber")
            .field("stream", &self.stream)
            .finish_non_exhaustive()
    }
}

impl KinesisSubscriber {
    /// The stream this subscription consumes from.
    #[must_use]
    pub fn stream_name(&self) -> &str {
        &self.stream
    }

    /// Opens the subscription and returns once it reads: every shard it took on the
    /// coordinator's first pass has a reader with its starting cursor, so a record published
    /// after this returns is one the subscription reads, and a seek made now reaches the reader
    /// of the shard it names.
    pub(crate) async fn open(core: &Arc<Core>, descriptor: KinesisStream) -> Self {
        let (tx, rx) = mpsc::channel(CHANNEL_CAPACITY);
        let stream = descriptor.stream().to_owned();
        let bus: SeekBus = Arc::new(SeekState::new(Arc::clone(core)));
        let limit = Arc::new(AtomicI32::new(DEFAULT_READ_LIMIT));
        let (reading, first_pass) = oneshot::channel();
        let coordinator = coordinate(
            Arc::new(Subscription {
                runtime: core.runtime.clone(),
                // The coordinator and its readers run on the runtime `connect` ran on, whose client
                // this is.
                client: core.client.home().clone(),
                store: Arc::clone(&core.store),
                owner: Arc::from(core.owner.as_str()),
                descriptor,
                out: tx,
                bus: Arc::clone(&bus),
                limit: Arc::clone(&limit),
            }),
            reading,
        );
        core.runtime.spawn(coordinator);
        // A coordinator that stops before its first pass drops the sender, which ends this wait
        // too; whatever stopped it is reported on the subscription's stream.
        let _ = first_pass.await;
        Self {
            stream,
            limit,
            deliveries: BufferedSubscriber::new(Deliveries::Shards(ShardDeliveries { rx, bus }))
                .max_wait(BATCH_FILL_WINDOW),
        }
    }

    /// A subscription on the in-process transport, batching on the client with the same deadline
    /// the service's subscription uses.
    #[cfg(feature = "testing")]
    pub(crate) fn in_process(stream: String, deliveries: LogDeliveries) -> Self {
        Self {
            stream,
            limit: Arc::new(AtomicI32::new(DEFAULT_READ_LIMIT)),
            deliveries: BufferedSubscriber::new(Deliveries::InProcess(deliveries))
                .max_wait(BATCH_FILL_WINDOW),
        }
    }
}

/// One delivery at a time, from whichever transport the broker was connected to: everything
/// above it batches on the client.
///
/// Without the `testing` feature the shard readers' channel is the only variant, so the type is
/// that channel and the `match` in [`stream`](Subscriber::stream) resolves at compile time.
enum Deliveries {
    Shards(ShardDeliveries),
    #[cfg(feature = "testing")]
    InProcess(LogDeliveries),
}

#[cfg(not(feature = "testing"))]
const _: () = assert!(size_of::<Deliveries>() == size_of::<ShardDeliveries>());

impl std::fmt::Debug for Deliveries {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Deliveries").finish_non_exhaustive()
    }
}

impl ruststream::Seekable for Deliveries {
    type Seeker = KinesisSeeker;

    fn seeker(&self) -> KinesisSeeker {
        match self {
            Self::Shards(shards) => ruststream::Seekable::seeker(shards),
            #[cfg(feature = "testing")]
            Self::InProcess(log) => log.seeker(),
        }
    }
}

impl Subscriber for Deliveries {
    type Message = KinesisMessage;
    type Error = KinesisError;

    fn stream(&mut self) -> impl Stream<Item = Result<KinesisMessage, KinesisError>> + Send + '_ {
        match self {
            #[cfg(not(feature = "testing"))]
            Self::Shards(shards) => shards.stream(),
            #[cfg(feature = "testing")]
            Self::Shards(shards) => Either::Left(shards.stream()),
            #[cfg(feature = "testing")]
            Self::InProcess(log) => Either::Right(log.stream()),
        }
    }
}

/// The subscription's raw delivery stream: the one channel every owned shard's reader feeds.
///
/// Kept apart from [`KinesisSubscriber`] so the framework's batch adapter can wrap it - the
/// adapter owns what it groups, and the public subscriber keeps its name and its stream-wide
/// surface.
struct ShardDeliveries {
    rx: mpsc::Receiver<Stamped>,
    bus: SeekBus,
}

impl std::fmt::Debug for ShardDeliveries {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ShardDeliveries").finish_non_exhaustive()
    }
}

/// The mount site's batch size as a `GetRecords` limit.
///
/// The service caps one read at 10000 records, so a larger batch is filled from several reads or
/// closes short when [`BATCH_FILL_WINDOW`] elapses: a batch carrying fewer records than the size
/// asked for is what the contract permits, one carrying more is not.
fn read_limit(size: NonZeroUsize) -> i32 {
    i32::try_from(size.get())
        .unwrap_or(MAX_READ_LIMIT)
        .min(MAX_READ_LIMIT)
}

/// Repositions a [`KinesisSubscriber`] while its stream runs; minted by
/// [`Seekable::seeker`](ruststream::Seekable::seeker).
///
/// A stream-wide position ([`KinesisPosition::Horizon`], [`Latest`](KinesisPosition::Latest),
/// [`Timestamp`](KinesisPosition::Timestamp)) moves every shard of the subscription and is
/// remembered, so shards whose readers start later open there too. A captured
/// [`Sequence`](KinesisPosition::Sequence) position moves the one shard it names. Either way
/// the affected shards drop their watermark bookkeeping: acknowledgements of records delivered
/// before the seek no longer checkpoint.
///
/// After the broker's `shutdown` every seek returns [`KinesisError::NotConnected`].
///
/// On a broker the test harness connected in process, the same handle re-reads that transport's
/// retained log instead, so a handler that repositions its subscription is the same code in a
/// test and against the service.
#[derive(Clone)]
pub struct KinesisSeeker {
    target: SeekTarget,
}

/// What this handle repositions. The two transports reposition different things, and an enum is
/// what keeps that from being a nullable field with only some combinations legal.
#[derive(Clone)]
enum SeekTarget {
    /// The service: one command channel per owned shard, behind a shared seek bus.
    Shards(ShardSeeker),
    /// The in-process transport: one retained log per stream.
    #[cfg(feature = "testing")]
    Log(LogSeeker),
}

#[cfg(not(feature = "testing"))]
const _: () = assert!(size_of::<SeekTarget>() == size_of::<ShardSeeker>());

impl std::fmt::Debug for KinesisSeeker {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("KinesisSeeker").finish_non_exhaustive()
    }
}

impl KinesisSeeker {
    /// Mints a handle onto one live subscription's seek surface: a reference-count bump, so the
    /// delivery path can carry one per record.
    pub(crate) fn shards(bus: SeekBus) -> Self {
        Self {
            target: SeekTarget::Shards(ShardSeeker { bus }),
        }
    }

    /// Mints a handle onto one in-process subscription's retained log.
    #[cfg(feature = "testing")]
    pub(crate) fn in_process(log: LogSeeker) -> Self {
        Self {
            target: SeekTarget::Log(log),
        }
    }
}

/// The service's half of [`KinesisSeeker`]: the shard readers a reposition is broadcast to.
#[derive(Clone)]
struct ShardSeeker {
    bus: SeekBus,
}

impl ShardSeeker {
    /// Repositions the one shard the captured position names.
    async fn seek_shard(&self, shard: String, start: ShardStart) -> Result<(), KinesisError> {
        let Some(handle) = self.bus.handle(&shard) else {
            return Err(KinesisError::Read {
                stream: String::new(),
                shard,
                source: Box::from("no live reader for this shard (not owned, or finished)"),
            });
        };
        // Bump the generation first: deliveries stamped before this instant are discarded,
        // including an in-flight batch the reader has not finished forwarding.
        handle.gate.fetch_add(1, Ordering::Release);
        let (done, wait) = oneshot::channel();
        handle
            .tx
            .send(ShardSeek { start, done })
            .map_err(|_| KinesisError::Read {
                stream: String::new(),
                shard: shard.clone(),
                source: Box::from("the shard reader has shut down"),
            })?;
        wait.await.map_err(|_| KinesisError::Read {
            stream: String::new(),
            shard,
            source: Box::from("the shard reader has shut down"),
        })?
    }

    /// Repositions every shard, present and future.
    async fn seek_stream(&self, start: StreamStart) -> Result<(), KinesisError> {
        // Installed before the broadcast, so a shard whose reader has not fetched its first
        // iterator yet lands on the position instead of racing past it. This is the whole
        // reason a `start_at(..)` clause works: it seeks the instant the subscription is
        // created, when no reader has started.
        self.bus.install(start);
        let handles = self.bus.live();
        // Every gate bumps before the first await, so a batch already in flight anywhere in
        // the subscription stamps stale.
        for handle in &handles {
            handle.gate.fetch_add(1, Ordering::Release);
        }
        let mut pending = Vec::with_capacity(handles.len());
        for handle in handles {
            let (done, wait) = oneshot::channel();
            // A reader that shut down between the snapshot and the send needs no reposition:
            // its shard is finished, or its successor will open at the installed position.
            if handle
                .tx
                .send(ShardSeek {
                    start: ShardStart::Stream(start),
                    done,
                })
                .is_ok()
            {
                pending.push(wait);
            }
        }
        for wait in pending {
            if let Ok(outcome) = wait.await {
                outcome?;
            }
        }
        Ok(())
    }
}

impl ShardSeeker {
    async fn seek(&self, to: KinesisPosition) -> Result<(), KinesisError> {
        self.bus.core.ensure_open()?;
        match to {
            KinesisPosition::Horizon => self.seek_stream(StreamStart::Horizon).await,
            KinesisPosition::Latest => self.seek_stream(StreamStart::Latest).await,
            KinesisPosition::Timestamp(millis) => {
                self.seek_stream(StreamStart::Timestamp(millis)).await
            }
            KinesisPosition::Sequence { shard, sequence } => {
                self.seek_shard(shard, ShardStart::At(sequence)).await
            }
        }
    }
}

impl ruststream::Seeker for KinesisSeeker {
    type Position = KinesisPosition;
    type Error = KinesisError;

    async fn seek(&self, to: KinesisPosition) -> Result<(), KinesisError> {
        match &self.target {
            SeekTarget::Shards(shards) => shards.seek(to).await,
            // Nothing here leaves the process, so the reposition resolves without awaiting.
            #[cfg(feature = "testing")]
            SeekTarget::Log(log) => log.seek(&to),
        }
    }
}

impl ruststream::Seekable for ShardDeliveries {
    type Seeker = KinesisSeeker;

    fn seeker(&self) -> KinesisSeeker {
        KinesisSeeker::shards(Arc::clone(&self.bus))
    }
}

impl Subscriber for ShardDeliveries {
    type Message = KinesisMessage;
    type Error = KinesisError;

    fn stream(&mut self) -> impl Stream<Item = Result<KinesisMessage, KinesisError>> + Send + '_ {
        // Poll the channel in place rather than wrapping it in an owning stream, so `stream`
        // can be called again after the returned stream is dropped (the runtime and the
        // conformance helpers re-enter it per call). Items stamped under an older generation
        // (before a seek) are discarded here.
        futures::stream::poll_fn(move |cx| {
            loop {
                match self.rx.poll_recv(cx) {
                    std::task::Poll::Ready(Some(stamped)) => {
                        if stamped.current() {
                            return std::task::Poll::Ready(Some(stamped.item));
                        }
                    }
                    std::task::Poll::Ready(None) => return std::task::Poll::Ready(None),
                    std::task::Poll::Pending => return std::task::Poll::Pending,
                }
            }
        })
    }
}

/// Repositioning reaches through the batch buffer to the readers, so a batch subscription opens
/// at a chosen position and repositions from a handler exactly as a single-message one does.
impl ruststream::Seekable for KinesisSubscriber {
    type Seeker = KinesisSeeker;

    fn seeker(&self) -> KinesisSeeker {
        ruststream::Seekable::seeker(&self.deliveries)
    }
}

impl Subscriber for KinesisSubscriber {
    type Message = KinesisMessage;
    type Error = KinesisError;

    fn stream(&mut self) -> impl Stream<Item = Result<KinesisMessage, KinesisError>> + Send + '_ {
        self.deliveries.stream()
    }
}

impl BatchSubscriber for KinesisSubscriber {
    type Batch = Vec<KinesisMessage>;

    /// # Cancel safety
    ///
    /// Cancel-safe between polls: a batch being filled lives inside the returned stream and
    /// survives a cancelled poll. Dropping the stream abandons that batch's records unsettled,
    /// so the shard replays them from its checkpoint when its lease is next taken.
    fn batches(
        &mut self,
        size: NonZeroUsize,
    ) -> impl Stream<Item = Result<Self::Batch, KinesisError>> + Send + '_ {
        // The batch size is the read limit: from the readers' next call on, one read fetches at
        // most one batch's worth. A read already in flight under the previous limit can still
        // return more, which is why the batch itself is capped below rather than here.
        self.limit.store(read_limit(size), Ordering::Relaxed);
        self.deliveries.batches(size)
    }
}

async fn list_all_shards(
    client: &aws_sdk_kinesis::Client,
    stream: &str,
) -> Result<Vec<Shard>, KinesisError> {
    // No paginator exists for ListShards, and a continuation call may carry ONLY the token.
    let mut shards = Vec::new();
    let mut token: Option<String> = None;
    loop {
        let request = token.take().map_or_else(
            || client.list_shards().stream_name(stream),
            |t| client.list_shards().next_token(t),
        );
        let output = request.send().await.map_err(|e| KinesisError::Stream {
            stream: stream.to_owned(),
            source: sdk_err(&e),
        })?;
        shards.extend(output.shards().iter().cloned());
        match output.next_token() {
            Some(t) => token = Some(t.to_owned()),
            None => return Ok(shards),
        }
    }
}

fn is_closed(shard: &Shard) -> bool {
    shard
        .sequence_number_range()
        .and_then(|r| r.ending_sequence_number())
        .is_some()
}

/// What one subscription's coordinator and every reader it spawns work from: one allocation per
/// subscription, one reference-count bump per reader.
struct Subscription {
    /// The runtime the broker connected on, where the coordinator starts every reader.
    runtime: Handle,
    client: aws_sdk_kinesis::Client,
    store: Arc<dyn LeaseStore>,
    /// This instance's lease-owner id, shared with every record a reader forwards.
    owner: Arc<str>,
    descriptor: KinesisStream,
    out: mpsc::Sender<Stamped>,
    bus: SeekBus,
    limit: Arc<AtomicI32>,
}

/// Runs one subscription's shard discovery for as long as the subscription is open.
///
/// `reading` is answered once the first pass is over and every reader it started has fetched its
/// starting cursor: until then a record published to a shard without a checkpoint could land
/// ahead of the tip the reader opens at, and be skipped. Errors of the first pass are reported
/// after that answer, since nothing reads the stream before `subscribe` returns.
async fn coordinate(subscription: Arc<Subscription>, reading: oneshot::Sender<()>) {
    let Subscription {
        runtime,
        client,
        store,
        owner,
        descriptor,
        out,
        bus,
        ..
    } = subscription.as_ref();
    let stream = descriptor.stream();
    // Leases are the stream's and the shard's together: shard ids repeat across streams.
    let stream_key: Arc<str> = Arc::from(stream);
    let mut readers: HashMap<String, tokio::task::JoinHandle<()>> = HashMap::new();
    let mut reading = Some(reading);
    // Filled on the first pass only: the readers' starting cursors, and the errors held back
    // until `subscribe` has returned.
    let mut positioned = Vec::new();
    let mut held = Vec::new();
    loop {
        readers.retain(|_, handle| !handle.is_finished());
        let first = reading.is_some();

        match list_all_shards(client, stream).await {
            Ok(shards) => {
                let by_id: HashMap<&str, &Shard> =
                    shards.iter().map(|s| (s.shard_id(), s)).collect();
                for shard in &shards {
                    let id = shard.shard_id().to_owned();
                    if readers.contains_key(&id) {
                        continue;
                    }
                    if !parents_done(bus, &stream_key, shard, &by_id, store.as_ref()).await {
                        continue;
                    }
                    let key = LeaseKey::new(Arc::clone(&stream_key), id.as_str());
                    match store.read(&key).await {
                        Ok(state) if state.checkpoint.as_deref() == Some(SHARD_END) => continue,
                        Ok(_) => {}
                        Err(err) => {
                            report(out, &mut held, first, lease_error(&key, err)).await;
                            continue;
                        }
                    }
                    match store.acquire(&key, owner, LEASE_TTL).await {
                        Ok(true) => {
                            let (seek_tx, seek_rx) = mpsc::unbounded_channel();
                            let handle = ShardHandle {
                                gate: Arc::new(AtomicU64::new(0)),
                                tx: seek_tx,
                            };
                            bus.register(id.clone(), handle.clone());
                            let started = first.then(|| {
                                let (started, cursor) = oneshot::channel();
                                positioned.push(cursor);
                                started
                            });
                            readers.insert(
                                id,
                                runtime.spawn(read_shard(
                                    Arc::clone(&subscription),
                                    Arc::new(key),
                                    handle.gate,
                                    seek_rx,
                                    started,
                                )),
                            );
                        }
                        Ok(false) => {} // another instance owns it
                        Err(err) => {
                            report(out, &mut held, first, lease_error(&key, err)).await;
                        }
                    }
                }
            }
            Err(err) => report(out, &mut held, first, err).await,
        }

        if let Some(reading) = reading.take() {
            for cursor in mem::take(&mut positioned) {
                // A reader that stopped before its cursor drops the sender; either way it is no
                // longer about to read.
                let _ = cursor.await;
            }
            let _ = reading.send(());
            for err in mem::take(&mut held) {
                let _ = out.send(Stamped::unstamped(Err(err))).await;
            }
        }

        tokio::select! {
            () = out.closed() => break,
            () = tokio::time::sleep(SHARD_SYNC) => {}
        }
    }
    // Readers watch the same channel and stop on their own.
}

/// Reports a coordinator error on the subscription's stream, or, on the first pass, holds it until
/// `subscribe` has returned: the stream has no reader before that, and a full channel would stall
/// the pass `subscribe` waits for.
async fn report(
    out: &mpsc::Sender<Stamped>,
    held: &mut Vec<KinesisError>,
    first: bool,
    err: KinesisError,
) {
    if first {
        held.push(err);
    } else {
        let _ = out.send(Stamped::unstamped(Err(err))).await;
    }
}

/// A child shard may start only when every parent is fully consumed - that is what keeps
/// per-key ordering across a split. A parent counts as done when it reached `SHARD_END`, was
/// trimmed out of the listing, or is closed with no checkpoint while the subscription starts
/// at the tip (its history is being skipped by request).
async fn parents_done(
    bus: &SeekState,
    stream: &Arc<str>,
    shard: &Shard,
    by_id: &HashMap<&str, &Shard>,
    store: &dyn LeaseStore,
) -> bool {
    let parents = [shard.parent_shard_id(), shard.adjacent_parent_shard_id()];
    for parent in parents.into_iter().flatten() {
        let Some(parent_shard) = by_id.get(parent) else {
            continue; // trimmed past retention
        };
        let Ok(state) = store.read(&LeaseKey::new(Arc::clone(stream), parent)).await else {
            return false;
        };
        match state.checkpoint.as_deref() {
            Some(SHARD_END) => {}
            None if is_closed(parent_shard)
                && matches!(bus.installed(), None | Some(StreamStart::Latest)) => {}
            _ => return false,
        }
    }
    true
}

/// Points a `GetShardIterator` request at a cursor.
fn iterator_at(
    request: GetShardIteratorFluentBuilder,
    start: &ShardStart,
) -> GetShardIteratorFluentBuilder {
    match start {
        ShardStart::Stream(StreamStart::Latest) => {
            request.shard_iterator_type(ShardIteratorType::Latest)
        }
        ShardStart::Stream(StreamStart::Horizon) => {
            request.shard_iterator_type(ShardIteratorType::TrimHorizon)
        }
        ShardStart::Stream(StreamStart::Timestamp(millis)) => request
            .shard_iterator_type(ShardIteratorType::AtTimestamp)
            .timestamp(DateTime::from_millis(
                i64::try_from(*millis).unwrap_or(i64::MAX),
            )),
        ShardStart::At(sequence) => request
            .shard_iterator_type(ShardIteratorType::AtSequenceNumber)
            .starting_sequence_number(sequence),
        ShardStart::After(sequence) => request
            .shard_iterator_type(ShardIteratorType::AfterSequenceNumber)
            .starting_sequence_number(sequence),
    }
}

/// Why a reader is fetching an iterator from scratch, which decides whether a sought position
/// or the stored checkpoint wins.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Reopen {
    /// The reader is starting. A position installed by a seek is forced here, ahead of the
    /// checkpoint: that is what the capability promises, and what `start_at(..)` means.
    Start,
    /// The reader lost its iterator mid-flight (expiry, a transient failure). The checkpoint
    /// is the truth now - re-applying the installed position would replay everything this
    /// reader has already handled - and the position only serves a shard that never
    /// checkpointed.
    Recover,
}

/// The crate's error for a lease store failure on `key`, naming the stream and the shard.
fn lease_error(key: &LeaseKey, source: LeaseError) -> KinesisError {
    KinesisError::Lease {
        stream: key.stream().to_owned(),
        shard: key.shard().to_owned(),
        source,
    }
}

async fn initial_iterator(
    client: &aws_sdk_kinesis::Client,
    key: &LeaseKey,
    bus: &SeekState,
    store: &dyn LeaseStore,
    why: Reopen,
) -> Result<Option<String>, KinesisError> {
    let (stream, shard) = (key.stream(), key.shard());
    let checkpoint = store
        .read(key)
        .await
        .map_err(|e| lease_error(key, e))?
        .checkpoint;
    if checkpoint.as_deref() == Some(SHARD_END) {
        return Ok(None);
    }
    let start = match (why, bus.installed(), checkpoint) {
        (Reopen::Start, Some(installed), _) | (Reopen::Recover, Some(installed), None) => {
            ShardStart::Stream(installed)
        }
        (_, _, Some(sequence)) => ShardStart::After(sequence),
        (_, None, None) => ShardStart::Stream(StreamStart::Latest),
    };
    let request = iterator_at(
        client
            .get_shard_iterator()
            .stream_name(stream)
            .shard_id(shard),
        &start,
    );
    let output = request.send().await.map_err(|e| KinesisError::Read {
        stream: stream.to_owned(),
        shard: shard.to_owned(),
        source: sdk_err(&e),
    })?;
    Ok(output.shard_iterator().map(str::to_owned))
}

/// Applies one reposition: a fresh iterator at the requested cursor and a reset watermark.
async fn apply_shard_seek(
    client: &aws_sdk_kinesis::Client,
    stream: &str,
    shard: &str,
    seek: ShardSeek,
    iterator: &mut String,
    tracker: &mut Arc<Watermark>,
) {
    let fresh = iterator_at(
        client
            .get_shard_iterator()
            .stream_name(stream)
            .shard_id(shard),
        &seek.start,
    )
    .send()
    .await;
    let outcome = match fresh {
        Ok(output) => output.shard_iterator().map_or_else(
            || {
                Err(KinesisError::Read {
                    stream: stream.to_owned(),
                    shard: shard.to_owned(),
                    source: Box::from("the service returned no iterator for the position"),
                })
            },
            |new_iterator| {
                new_iterator.clone_into(iterator);
                *tracker = Arc::new(Watermark::default());
                Ok(())
            },
        ),
        Err(err) => Err(KinesisError::Read {
            stream: stream.to_owned(),
            shard: shard.to_owned(),
            source: sdk_err(&err),
        }),
    };
    let _ = seek.done.send(outcome);
}

#[allow(clippy::too_many_lines)]
async fn read_shard(
    subscription: Arc<Subscription>,
    lease: Arc<LeaseKey>,
    gate: Arc<AtomicU64>,
    mut seek_rx: mpsc::UnboundedReceiver<ShardSeek>,
    started: Option<oneshot::Sender<()>>,
) {
    // Every exit path must deregister this shard's seek surface.
    struct BusGuard {
        bus: SeekBus,
        shard: Arc<str>,
    }
    impl Drop for BusGuard {
        fn drop(&mut self) {
            self.bus.deregister(&self.shard);
        }
    }
    let Subscription {
        client,
        store,
        owner,
        descriptor,
        out,
        bus,
        limit,
        ..
    } = subscription.as_ref();
    let shard = lease.shard_shared();
    let _bus_guard = BusGuard {
        bus: Arc::clone(bus),
        shard: Arc::clone(shard),
    };
    // Minted once, before the first delivery: every record carries a clone so a handler can
    // reposition the subscription from its per-delivery context.
    let seeker = KinesisSeeker::shards(Arc::clone(bus));
    let stream = descriptor.stream();
    let mut tracker = Arc::new(Watermark::default());
    let initial = initial_iterator(client, &lease, bus, store.as_ref(), Reopen::Start).await;
    // The cursor is fixed, so the tip this shard opens at is the one `subscribe` promised.
    drop(started);
    let mut iterator = match initial {
        Ok(Some(iterator)) => iterator,
        Ok(None) => return, // already at SHARD_END
        Err(err) => {
            let _ = out.send(Stamped::unstamped(Err(err))).await;
            return;
        }
    };
    let mut last_renew = tokio::time::Instant::now();
    let mut failures: u32 = 0;

    loop {
        if out.is_closed() {
            let _ = store.release(&lease, owner).await;
            return;
        }
        // A reposition replaces the iterator and resets the watermark; deliveries stamped
        // under the previous generation are discarded by their settlements and were already
        // filtered from checkpointing by the gate bump at enqueue.
        while let Ok(seek) = seek_rx.try_recv() {
            apply_shard_seek(client, stream, shard, seek, &mut iterator, &mut tracker).await;
        }
        if last_renew.elapsed() >= RENEW_EVERY {
            match store.renew(&lease, owner, LEASE_TTL).await {
                Ok(true) => last_renew = tokio::time::Instant::now(),
                // Fenced: stop immediately, without checkpointing - the new owner replays
                // from the last checkpoint, which at-least-once permits.
                Ok(false) | Err(_) => return,
            }
        }

        // Stamped before the read: a seek that lands while the batch is in flight bumps the
        // gate, so these deliveries are discarded rather than leaking pre-seek records.
        let epoch = gate.load(Ordering::Acquire);
        let response = client
            .get_records()
            .shard_iterator(&iterator)
            .limit(limit.load(Ordering::Relaxed))
            .send()
            .await;
        match response {
            Ok(output) => {
                failures = 0;
                for record in output.records() {
                    let data = record.data().as_ref();
                    if data.len() > 4 && data[0..4] == KPL_MAGIC {
                        if out
                            .send(Stamped::live(
                                epoch,
                                &gate,
                                Err(KinesisError::AggregatedRecord {
                                    shard: shard.to_string(),
                                }),
                            ))
                            .await
                            .is_err()
                        {
                            return;
                        }
                        continue;
                    }
                    let settlement = Settle::Checkpoint(Settlement {
                        tracker: Arc::clone(&tracker),
                        index: tracker.deliver(record.sequence_number()),
                        store: Arc::clone(store),
                        lease: Arc::clone(&lease),
                        owner: Arc::clone(owner),
                        epoch,
                        gate: Arc::clone(&gate),
                    });
                    let message = KinesisMessage::new(
                        data,
                        // The service stamps every record with the key it was put under; the
                        // SDK's model only types the field as optional.
                        record.partition_key().unwrap_or_default(),
                        record.sequence_number(),
                        seeker.clone(),
                        settlement,
                    );
                    if out
                        .send(Stamped::live(epoch, &gate, Ok(message)))
                        .await
                        .is_err()
                    {
                        let _ = store.release(&lease, owner).await;
                        return;
                    }
                }

                let Some(next) = output.next_shard_iterator() else {
                    // SHARD_END. Wait for every delivery to settle, then mark the shard
                    // finished so the coordinator may start its children.
                    while !tracker.drained() && !out.is_closed() {
                        tokio::time::sleep(Duration::from_millis(100)).await;
                    }
                    if tracker.drained() {
                        let _ = store.checkpoint(&lease, owner, SHARD_END).await;
                    }
                    let _ = store.release(&lease, owner).await;
                    return;
                };
                iterator = next.to_owned();
                if output.records().is_empty() {
                    tokio::select! {
                        () = out.closed() => {}
                        seek = seek_rx.recv() => {
                            if let Some(seek) = seek {
                                apply_shard_seek(
                                    client, stream, shard, seek, &mut iterator, &mut tracker,
                                )
                                .await;
                            }
                        }
                        () = tokio::time::sleep(descriptor.poll_value()) => {}
                    }
                }
            }
            Err(err) => {
                let expired = err
                    .as_service_error()
                    .is_some_and(GetRecordsError::is_expired_iterator_exception);
                let fatal = err
                    .as_service_error()
                    .is_some_and(GetRecordsError::is_resource_not_found_exception);
                // Throttling is the service asking for a slower cadence, not a failed read:
                // the iterator is still good, so the reader waits and reuses it. Reporting it
                // would put an error in front of a handler that did nothing wrong, and the
                // recovery below would rewind the shard to its checkpoint and redeliver
                // everything since. A batch smaller than a read's natural size makes this the
                // common answer to a backlog, so it must not cost either.
                if err
                    .as_service_error()
                    .is_some_and(GetRecordsError::is_provisioned_throughput_exceeded_exception)
                {
                    tokio::time::sleep(descriptor.poll_value()).await;
                    continue;
                }
                if !expired
                    && out
                        .send(Stamped::unstamped(Err(KinesisError::Read {
                            stream: stream.to_owned(),
                            shard: shard.to_string(),
                            source: sdk_err(&err),
                        })))
                        .await
                        .is_err()
                {
                    return;
                }
                if fatal {
                    let _ = store.release(&lease, owner).await;
                    return;
                }
                failures += 1;
                if failures > 10 {
                    let _ = store.release(&lease, owner).await;
                    return;
                }
                // Expired iterators are refetched from the checkpoint (never Latest, which
                // would silently skip data); everything else backs off and retries.
                tokio::time::sleep(Duration::from_secs(1)).await;
                match initial_iterator(client, &lease, bus, store.as_ref(), Reopen::Recover).await {
                    Ok(Some(fresh)) => iterator = fresh,
                    Ok(None) => return,
                    Err(_) => {}
                }
            }
        }
    }
}