skilj-core 0.2.0

//! Group-commit batching for `db::submit_command` (issue #32,
//! round two). The first pass at this issue (docs/architecture.md §57)
//! shortened how long `submit_command`'s own bounded-context lock is
//! held per command; a real load test against `skilj-helpdesk` still
//! found the ceiling this exists to raise, because shortening one
//! command's own critical section does nothing about the fundamental
//! shape of the problem: every command commit to one bounded context is
//! *fully serialized* through that one `sequence` row's lock (required for
//! `SequenceIsGaplessPerBoundedContext`/`DynamicConsistencyBoundaryHonoured`,
//! not something this pass touches), so under real concurrent load the
//! throughput ceiling is `1 / (lock hold time)`, full stop, no matter how
//! short each individual hold is.
//!
//! **The actual lever left: make one lock acquisition serve more than
//! one command.** `CommandBatcher` implements a "group commit" -
//! standard database terminology for the same pattern write-ahead-log
//! flushes and MySQL's binlog use: instead of every concurrent caller
//! independently opening its own transaction and queueing for the lock,
//! callers that arrive close together in time are coalesced into one
//! shared batch, processed by whichever one of them becomes that batch's
//! *leader* inside a single transaction (`db::submit_command_batch`,
//! each command isolated by its own `SAVEPOINT` so one command's real
//! failure never affects another's), committed once. N queued callers
//! that used to mean N separate lock acquisitions now mean one.
//!
//! **Self-tuning, no batch-window or batch-size knob to guess**: a
//! caller becomes the leader for a new batch only when it finds the
//! per-bounded-context queue empty at the moment it pushes onto it
//! (`submit`'s own `is_leader` check, guarded by the queue's own
//! `Mutex`, so exactly one caller can ever observe that transition per
//! batch). The leader does *not* immediately grab whatever's in the
//! queue - it first does the slow part (`pool.begin()` plus the actual
//! `SELECT ... FOR UPDATE` wait, which is exactly the time other
//! commands are still arriving and queueing behind it under real
//! contention) and only *then* drains the queue, taking everyone who
//! joined in the meantime. Under low load, a lone caller becomes its own
//! leader, finds nobody else queued once its lock is acquired, and pays
//! essentially the identical latency `submit_command` alone would have
//! (batch of one). Under heavy load - exactly the case the load test
//! found collapsing - the leader's own wait for the lock is naturally
//! longer (every earlier batch is still ahead of it), so more callers
//! accumulate in that same window, producing a *larger* batch precisely
//! when amortising the lock acquisition matters most. No artificial
//! sleep, no fixed batch-size cap needed for this to self-tune (though
//! the configured max batch size below still bounds the pathological case).
//!
//! **Correctness, not just throughput**: every invariant `submit_command`
//! itself guarantees still holds, unweakened - see `db::submit_command_batch`'s
//! own doc comment for exactly how (per-command `SAVEPOINT` isolation,
//! `extra_committed_events` giving same-batch commands the identical DCB
//! conflict visibility a truly separate transaction would have had via
//! the lock alone, replies only ever sent after the shared `tx.commit()`
//! actually succeeds).

use crate::db::{
    BatchedCommand, BatchedCommandResult, OwnedSnapshotContext, Pool, SnapshotContext,
    SubmitCommandOutcome,
};
use crate::encryption::KeyService;
use crate::error::{Error, Result};
use crate::event_cache::EventCache;
use crate::event_store::{CommandType, Event, EventBroadcaster};
use crate::plugin::{CommandDispatcher, ProjectionDispatcher};
use crate::shared::{CommandDecision, Tag};
use chrono::{DateTime, Utc};
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::Mutex;
use tokio::sync::{oneshot, RwLock, Semaphore};
use tracing::Instrument;

/// The default for [`CommandBatcher::with_max_batch_size`]. A hard ceiling on how many commands one lock acquisition processes,
/// regardless of how many piled up while the leader waited for the
/// lock - a pathological worst case (thousands of callers queued behind
/// one very slow commit) still bounds one batch's own lock hold time and
/// memory, at the cost of spilling the remainder into a follow-up batch
/// (the same leader keeps draining and processing follow-up batches until
/// the queue is empty - see `run_as_leader` - so nothing is lost or
/// dropped, just processed one batch later).
pub const DEFAULT_MAX_BATCH_SIZE: usize = 256;

/// How many commands of one [`CommandBatcher::decide_and_submit_many`]
/// call read their history and run `decide()` at once before they are
/// queued. Each takes a pooled connection for its read, so this keeps one
/// large call from taking the whole pool.
pub const PREPARE_CONCURRENCY: usize = 8;

/// The default for [`CommandBatcher::with_max_items_per_batch`] -
/// `config.max_items_per_batch` in specs/skilj.allium.
pub const DEFAULT_MAX_ITEMS_PER_BATCH: usize = 100;

/// One command of a [`CommandBatcher::decide_and_submit_many`] call: what
/// a single [`CommandBatcher::decide_and_submit`] takes for one command,
/// already authorised by the caller.
#[derive(Debug, Clone)]
pub struct CommandBatchItem {
    /// The command's type; every item of one call belongs to the call's
    /// bounded context.
    pub command_type: CommandType,
    /// The payload, as JSON.
    pub payload: String,
    /// The submitting caller - a Role's or a token's id.
    pub client_id: String,
    /// The caller's correlation id, or `None` to have one generated.
    pub correlation_id: Option<String>,
    /// The caller's causation id, if any.
    pub causation_id: Option<String>,
    /// The caller's idempotency key, if any.
    pub idempotency_key: Option<String>,
}

/// One command, sitting in a per-bounded-context queue - see this
/// module's own doc comment. Every entry is answered through its `reply`
/// sender, the leader's own included: since a caller can queue several
/// commands at once (`decide_and_submit_many`, docs/architecture.md
/// §227), "the leader's command" is no longer one entry the leader can
/// read its answer off directly.
struct PendingCommand {
    command: BatchedCommand,
    reply: oneshot::Sender<BatchedCommandResult>,
}

type Queue = Arc<Mutex<Vec<PendingCommand>>>;

/// The default for [`CommandBatcher::with_idle_in_transaction_timeout`] -
/// see that method's own doc comment. 30 seconds is generous for
/// everything a batch leader's own transaction legitimately does (one row
/// lock, a handful of metadata reads, up to the configured max batch size commands'
/// worth of decider/projection work, all of it CPU-bound or against the
/// same already-warm database), so tripping it is always a real stall,
/// never an ordinary slow batch.
const DEFAULT_IDLE_IN_TRANSACTION_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);

/// See this module's own doc comment for the full design. `Clone` is
/// cheap - `Arc`-wrapped internals, the identical "hand out a cheap
/// clone, not an `Arc<Skilj>`" reasoning `EventCache`/`EventBroadcaster`
/// already use.
#[derive(Clone)]
pub struct CommandBatcher {
    queues: Arc<RwLock<HashMap<String, Queue>>>,
    idle_in_transaction_timeout: std::time::Duration,
    /// Caps how many batch leaders (each pinning one pool connection for
    /// its whole batch while still needing *more* pool connections for
    /// its decide/persist reads) can run at once across every bounded
    /// context, sized lazily from the pool's own `max_connections` - see
    /// `leader_permits`.
    leader_permits: Arc<std::sync::OnceLock<Arc<Semaphore>>>,
    max_batch_size: usize,
    max_concurrent_leaders: Option<usize>,
    max_items_per_batch: usize,
    slow: crate::slow_command::SlowCommandWarnings,
}

impl Default for CommandBatcher {
    fn default() -> Self {
        Self::new()
    }
}

impl CommandBatcher {
    pub fn new() -> Self {
        Self {
            queues: Arc::new(RwLock::new(HashMap::new())),
            idle_in_transaction_timeout: DEFAULT_IDLE_IN_TRANSACTION_TIMEOUT,
            leader_permits: Arc::new(std::sync::OnceLock::new()),
            max_batch_size: DEFAULT_MAX_BATCH_SIZE,
            max_concurrent_leaders: None,
            max_items_per_batch: DEFAULT_MAX_ITEMS_PER_BATCH,
            slow: crate::slow_command::SlowCommandWarnings::default(),
        }
    }

    /// How long a command's read before the lock may take before a
    /// warning is logged (default
    /// [`DEFAULT_SLOW_READ_WARNING`](crate::slow_command::DEFAULT_SLOW_READ_WARNING));
    /// `None` never warns. See docs/architecture.md §203.
    pub fn with_slow_read_warning(mut self, threshold: Option<std::time::Duration>) -> Self {
        self.slow = self.slow.with_read(threshold);
        self
    }

    /// How long one `decide()` may take, before the lock or under it,
    /// before a warning is logged (default
    /// [`DEFAULT_SLOW_DECIDE_WARNING`](crate::slow_command::DEFAULT_SLOW_DECIDE_WARNING));
    /// `None` never warns. See docs/architecture.md §203.
    pub fn with_slow_decide_warning(mut self, threshold: Option<std::time::Duration>) -> Self {
        self.slow = self.slow.with_decide(threshold);
        self
    }

    /// The most commands one lock acquisition processes (default
    /// [`DEFAULT_MAX_BATCH_SIZE`]). A larger value amortizes the lock and
    /// commit over more commands; a smaller one bounds how long one batch
    /// holds the bounded-context lock and how much work a single failed
    /// batch takes down with it. Values below 1 are treated as 1.
    pub fn with_max_batch_size(mut self, max_batch_size: usize) -> Self {
        self.max_batch_size = max_batch_size.max(1);
        self
    }

    /// The most items one batch submission carries - commands to
    /// `submitCommands` or the REST batch trigger, or external events to
    /// the REST batch route (docs/architecture.md §227). A call with more
    /// is refused whole. Kept here, not on the routes, because both
    /// surfaces already share this batcher. Clamped to at least 1.
    pub fn with_max_items_per_batch(mut self, max_items: usize) -> Self {
        self.max_items_per_batch = max_items.max(1);
        self
    }

    /// See [`with_max_items_per_batch`](Self::with_max_items_per_batch).
    pub fn max_items_per_batch(&self) -> usize {
        self.max_items_per_batch
    }

    /// The most batch leaders that may run at once across all bounded
    /// contexts (each pins one pool connection for its whole batch).
    /// Defaults to half the pool's `max_connections` (minimum 1) - see
    /// `leader_permits`. Values below 1 are treated as 1. Must be set
    /// before the first command is submitted.
    pub fn with_max_concurrent_leaders(mut self, max_concurrent_leaders: usize) -> Self {
        self.max_concurrent_leaders = Some(max_concurrent_leaders.max(1));
        self
    }

    /// At most half the pool (minimum one) may be pinned by batch leaders
    /// at once, so every other caller's warm-up/optimistic resolve can
    /// still get a connection while leaders hold theirs. A leader itself
    /// needs no second connection under its lock - everything it reads
    /// or provisions there runs on its own transaction
    /// (docs/architecture.md §117).
    fn leader_permits(&self, pool: &Pool) -> Arc<Semaphore> {
        self.leader_permits
            .get_or_init(|| {
                let permits = self
                    .max_concurrent_leaders
                    .unwrap_or((pool.options().get_max_connections() as usize / 2).max(1));
                Arc::new(Semaphore::new(permits))
            })
            .clone()
    }

    /// issue #36's own recommendation #3: a defense-in-depth
    /// backstop against a batch leader's transaction stalling - for any
    /// reason, root cause fixed or not - while holding the bounded-context
    /// lock. Issued as `SET LOCAL idle_in_transaction_session_timeout` on
    /// the leader's own transaction (see `db::begin_command_batch_leader_tx`'s
    /// own doc comment), so Postgres itself kills a genuinely stuck leader
    /// and releases the lock instead of every writer to that bounded
    /// context queueing forever behind a wedge nothing else ever clears.
    pub fn with_idle_in_transaction_timeout(mut self, timeout: std::time::Duration) -> Self {
        self.idle_in_transaction_timeout = timeout;
        self
    }

    /// The `Queue` for `bounded_context`, creating it (empty) on first
    /// use - a read-mostly lookup (`RwLock::read`) for every bounded
    /// context this process has already touched at least once, falling
    /// back to a brief write lock only the first time.
    async fn queue_for(&self, bounded_context: &str) -> Queue {
        if let Some(queue) = self.queues.read().await.get(bounded_context) {
            return queue.clone();
        }
        self.queues
            .write()
            .await
            .entry(bounded_context.to_string())
            .or_insert_with(|| Arc::new(Mutex::new(Vec::new())))
            .clone()
    }

    /// The batched drop-in replacement for `db::submit_command` - same
    /// contract (identical `SubmitCommandOutcome`, identical set of
    /// error cases a single unbatched call could produce), just routed
    /// through this bounded context's own shared queue instead of always
    /// opening its own transaction. See this module's own doc comment
    /// for why joining that queue never changes what a caller observes,
    /// only how many other commands happen to share the lock acquisition
    /// that produces its result.
    #[allow(clippy::too_many_arguments)]
    pub async fn submit(
        &self,
        pool: &Pool,
        dispatcher: &Arc<dyn CommandDispatcher>,
        projection_dispatcher: &Arc<dyn ProjectionDispatcher>,
        broadcaster: &EventBroadcaster,
        event_cache: &EventCache,
        command_type: &CommandType,
        payload: &str,
        client_id: &str,
        correlation_id: Option<&str>,
        causation_id: Option<&str>,
        bounded_context_events: &[Event],
        consistency_tags: &[Tag],
        matching_events: &[Event],
        initial_decision: CommandDecision,
        key_service: Option<&KeyService>,
        now: DateTime<Utc>,
        snapshot: Option<SnapshotContext<'_>>,
        snapshot_states: &[crate::plugin::SnapshotState],
        idempotency_key: Option<&str>,
        covered_through: Option<i64>,
    ) -> Result<SubmitCommandOutcome> {
        let command = Self::prepare(
            pool,
            command_type,
            payload,
            client_id,
            correlation_id,
            causation_id,
            bounded_context_events,
            consistency_tags,
            matching_events,
            initial_decision,
            key_service,
            now,
            snapshot,
            snapshot_states,
            idempotency_key,
            covered_through,
        )
        .await?;
        let reply = self
            .enqueue(
                pool,
                dispatcher,
                projection_dispatcher,
                broadcaster,
                event_cache,
                key_service,
                &command_type.bounded_context.name,
                vec![command],
            )
            .await
            .pop()
            .expect("one command queued, one reply");
        await_reply(reply).await
    }

    /// A command as it waits in the queue. The pre-lock warm-up is done
    /// here - fully in parallel across every concurrently-submitting
    /// caller, before any of them join the shared queue - rather than by
    /// whichever one becomes leader, serially, for the whole batch. See
    /// `db::warm_up_event_types_and_encryption_keys`'s own doc comment for
    /// why this is always safe.
    #[allow(clippy::too_many_arguments)]
    async fn prepare(
        pool: &Pool,
        command_type: &CommandType,
        payload: &str,
        client_id: &str,
        correlation_id: Option<&str>,
        causation_id: Option<&str>,
        bounded_context_events: &[Event],
        consistency_tags: &[Tag],
        matching_events: &[Event],
        initial_decision: CommandDecision,
        key_service: Option<&KeyService>,
        now: DateTime<Utc>,
        snapshot: Option<SnapshotContext<'_>>,
        snapshot_states: &[crate::plugin::SnapshotState],
        idempotency_key: Option<&str>,
        covered_through: Option<i64>,
    ) -> Result<BatchedCommand> {
        let (event_types_by_name, resolved) = crate::db::warm_up_event_types_and_encryption_keys(
            pool,
            &command_type.bounded_context.name,
            command_type,
            payload,
            &initial_decision,
            key_service,
        )
        .await?;

        Ok(BatchedCommand {
            command_type: command_type.clone(),
            payload: payload.to_string(),
            client_id: client_id.to_string(),
            correlation_id: correlation_id.map(str::to_string),
            causation_id: causation_id.map(str::to_string),
            bounded_context_events: bounded_context_events.to_vec(),
            covered_through,
            consistency_tags: consistency_tags.to_vec(),
            matching_events: matching_events.to_vec(),
            initial_decision,
            now,
            snapshot: snapshot.map(|s| OwnedSnapshotContext {
                state_json: s.state_json.to_string(),
                as_of_sequence: s.as_of_sequence,
            }),
            snapshot_states: snapshot_states.to_vec(),
            idempotency_key: idempotency_key.map(str::to_string),
            event_types_by_name,
            resolved,
        })
    }

    /// Queues `commands`, in order, in one push - so nothing another
    /// caller queues lands between them in the queue - and becomes the
    /// leader if the queue was empty. One reply per command, in the same
    /// order.
    #[allow(clippy::too_many_arguments)]
    async fn enqueue(
        &self,
        pool: &Pool,
        dispatcher: &Arc<dyn CommandDispatcher>,
        projection_dispatcher: &Arc<dyn ProjectionDispatcher>,
        broadcaster: &EventBroadcaster,
        event_cache: &EventCache,
        key_service: Option<&KeyService>,
        bounded_context_name: &str,
        commands: Vec<BatchedCommand>,
    ) -> Vec<oneshot::Receiver<BatchedCommandResult>> {
        if commands.is_empty() {
            return Vec::new();
        }
        let queue = self.queue_for(bounded_context_name).await;
        let count = commands.len();
        let mut replies = Vec::with_capacity(count);
        let is_leader = {
            let mut queue = lock_queue(&queue);
            for command in commands {
                let (reply, reply_rx) = oneshot::channel();
                queue.push(PendingCommand { command, reply });
                replies.push(reply_rx);
            }
            // Exactly one pusher can ever observe the queue going from
            // empty to non-empty - the push and this check happen under
            // the same lock acquisition, so leadership for a given batch
            // is assigned unambiguously, exactly once.
            queue.len() == count
        };

        if is_leader {
            // The batch runs on its own task, not on this caller's future:
            // a leader whose client disconnects (axum drops the handler
            // future) would otherwise drop the shared transaction
            // mid-batch, failing every batch-mate's command - or, after the
            // commit, leaving them told `BatchFailed` for a command that
            // did commit, and its events unbroadcast (docs/architecture.md
            // §114). Detached, the batch finishes and answers everyone; the
            // leader's own command commits even if nobody is left to read
            // its result, as any command already on its way into a
            // transaction can.
            let leader = self.clone();
            let pool = pool.clone();
            let dispatcher = dispatcher.clone();
            let projection_dispatcher = projection_dispatcher.clone();
            let broadcaster = broadcaster.clone();
            let event_cache = event_cache.clone();
            let key_service = key_service.cloned();
            let bounded_context_name = bounded_context_name.to_string();
            tokio::spawn(
                async move {
                    leader
                        .run_as_leader(
                            &pool,
                            dispatcher.as_ref(),
                            projection_dispatcher.as_ref(),
                            &broadcaster,
                            &event_cache,
                            key_service.as_ref(),
                            &bounded_context_name,
                            &queue,
                        )
                        .await
                }
                .instrument(tracing::Span::current()),
            );
        }
        replies
    }

    /// Several commands for one bounded context, decided and committed in
    /// the order given, each with the outcome it would have had sent alone
    /// right after the one before it - `rule SubmitCommandBatch`
    /// (docs/architecture.md §227). One result per command, in order: a
    /// command that fails before it is queued (its read, its decide, its
    /// warm-up) fails alone, and the others are still submitted.
    ///
    /// Each command's optimistic read and `decide()` run before any is
    /// queued, up to [`PREPARE_CONCURRENCY`] at a time, so a later command
    /// doesn't see an earlier one's events there. It does under the lock:
    /// the batch's commands are queued in one push, in order, and the
    /// leader decides each against the events of the commands before it
    /// (`extra_committed_events`), deciding again when they matter.
    ///
    /// With `stop_at_first_failure`, the commands after the first one
    /// that fails before it is queued are not queued at all and answer
    /// [`Error::NotProcessed`], so they can be sent again in order. A
    /// command that fails under the lock - a database error in its
    /// savepoint, a sync projection refusing its event - doesn't stop the
    /// commands queued with it, which the leader has decided by then.
    ///
    /// Unlike [`decide_and_submit`](Self::decide_and_submit), a command
    /// whose blind tags a key rotation made stale (§209) is answered with
    /// that error rather than submitted again: submitted again, it would
    /// land after the commands that followed it.
    #[allow(clippy::too_many_arguments)]
    pub async fn decide_and_submit_many(
        &self,
        pool: &Pool,
        dispatcher: &Arc<dyn CommandDispatcher>,
        projection_dispatcher: &Arc<dyn ProjectionDispatcher>,
        snapshot_dispatcher: &dyn crate::plugin::SnapshotDispatcher,
        broadcaster: &EventBroadcaster,
        event_cache: &EventCache,
        key_service: Option<&KeyService>,
        bounded_context_name: &str,
        commands: Vec<CommandBatchItem>,
        stop_at_first_failure: bool,
    ) -> Vec<Result<SubmitCommandOutcome>> {
        use futures_util::StreamExt;

        let prepared: Vec<Result<BatchedCommand>> = futures_util::stream::iter(commands)
            .map(|item| async move {
                if item.command_type.bounded_context.name != bounded_context_name {
                    return Err(crate::access_control::Error::GrantBoundedContextMismatch.into());
                }
                let now = Utc::now();
                let resolved = crate::db::resolve_command_submission(
                    pool,
                    dispatcher.as_ref(),
                    snapshot_dispatcher,
                    event_cache,
                    &item.command_type,
                    &item.payload,
                    &crate::plugin::DecisionContext::at(now),
                )
                .await?;
                self.slow.note_resolved(&item.command_type, &resolved);
                Self::prepare(
                    pool,
                    &item.command_type,
                    &item.payload,
                    &item.client_id,
                    item.correlation_id.as_deref(),
                    item.causation_id.as_deref(),
                    &resolved.bounded_context_events,
                    &resolved.consistency_tags,
                    &resolved.matching_events,
                    resolved.decision,
                    key_service,
                    now,
                    resolved
                        .snapshot_context
                        .as_ref()
                        .map(|ctx| SnapshotContext {
                            state_json: &ctx.state_json,
                            as_of_sequence: ctx.as_of_sequence,
                        }),
                    &resolved.snapshot_states,
                    item.idempotency_key.as_deref(),
                    Some(resolved.covered_through),
                )
                .await
            })
            .buffered(PREPARE_CONCURRENCY)
            .collect()
            .await;

        let mut queued = Vec::new();
        let mut failed = false;
        let mut results: Vec<Option<Result<SubmitCommandOutcome>>> = prepared
            .into_iter()
            .map(|command| match command {
                Ok(_) if failed && stop_at_first_failure => Some(Err(Error::NotProcessed)),
                Ok(command) => {
                    queued.push(command);
                    None
                }
                Err(e) => {
                    failed = true;
                    Some(Err(e))
                }
            })
            .collect();
        let mut replies = self
            .enqueue(
                pool,
                dispatcher,
                projection_dispatcher,
                broadcaster,
                event_cache,
                key_service,
                bounded_context_name,
                queued,
            )
            .await
            .into_iter();
        for result in results.iter_mut().filter(|r| r.is_none()) {
            let reply = replies.next().expect("one reply per queued command");
            *result = Some(await_reply(reply).await);
        }
        results
            .into_iter()
            .map(|r| r.expect("every command has a result"))
            .collect()
    }

    /// The batched drop-in replacement for `db::decide_and_submit_command` -
    /// same optimistic-resolve-then-submit shape
    /// (`db::resolve_command_submission` computes the identical optimistic
    /// half either way), just handing its result to this batcher's own
    /// [`submit`](Self::submit) instead of `db::submit_command` directly.
    /// The one real entry point REST/GraphQL command submission and
    /// parked-delivery redrive are meant to call - see `db::decide_and_submit_command`'s
    /// own doc comment for why the lower-volume, periodic internal
    /// callers (the cross-context router's tick, `fire_due_deadlines`)
    /// deliberately still call that one directly instead.
    #[allow(clippy::too_many_arguments)]
    pub async fn decide_and_submit(
        &self,
        pool: &Pool,
        dispatcher: &Arc<dyn CommandDispatcher>,
        projection_dispatcher: &Arc<dyn ProjectionDispatcher>,
        snapshot_dispatcher: &dyn crate::plugin::SnapshotDispatcher,
        broadcaster: &EventBroadcaster,
        event_cache: &EventCache,
        command_type: &CommandType,
        payload: &str,
        client_id: &str,
        correlation_id: Option<&str>,
        causation_id: Option<&str>,
        key_service: Option<&KeyService>,
        now: DateTime<Utc>,
        idempotency_key: Option<&str>,
    ) -> Result<SubmitCommandOutcome> {
        // docs/architecture.md §209: once more, rehashed, when a key
        // rotation committed between deriving the tags and the lock.
        let mut rotated_once = false;
        loop {
            let outcome = self
                .decide_and_submit_once(
                    pool,
                    dispatcher,
                    projection_dispatcher,
                    snapshot_dispatcher,
                    broadcaster,
                    event_cache,
                    command_type,
                    payload,
                    client_id,
                    correlation_id,
                    causation_id,
                    key_service,
                    now,
                    idempotency_key,
                )
                .await;
            match outcome {
                Err(err) if crate::db::blind_tags_rotated(&err) && !rotated_once => {
                    rotated_once = true
                }
                outcome => return outcome,
            }
        }
    }

    #[allow(clippy::too_many_arguments)]
    async fn decide_and_submit_once(
        &self,
        pool: &Pool,
        dispatcher: &Arc<dyn CommandDispatcher>,
        projection_dispatcher: &Arc<dyn ProjectionDispatcher>,
        snapshot_dispatcher: &dyn crate::plugin::SnapshotDispatcher,
        broadcaster: &EventBroadcaster,
        event_cache: &EventCache,
        command_type: &CommandType,
        payload: &str,
        client_id: &str,
        correlation_id: Option<&str>,
        causation_id: Option<&str>,
        key_service: Option<&KeyService>,
        now: DateTime<Utc>,
        idempotency_key: Option<&str>,
    ) -> Result<SubmitCommandOutcome> {
        let resolved = crate::db::resolve_command_submission(
            pool,
            dispatcher.as_ref(),
            snapshot_dispatcher,
            event_cache,
            command_type,
            payload,
            &crate::plugin::DecisionContext::at(now),
        )
        .await?;
        self.slow.note_resolved(command_type, &resolved);

        self.submit(
            pool,
            dispatcher,
            projection_dispatcher,
            broadcaster,
            event_cache,
            command_type,
            payload,
            client_id,
            correlation_id,
            causation_id,
            &resolved.bounded_context_events,
            &resolved.consistency_tags,
            &resolved.matching_events,
            resolved.decision,
            key_service,
            now,
            resolved
                .snapshot_context
                .as_ref()
                .map(|ctx| SnapshotContext {
                    state_json: &ctx.state_json,
                    as_of_sequence: ctx.as_of_sequence,
                }),
            &resolved.snapshot_states,
            idempotency_key,
            Some(resolved.covered_through),
        )
        .await
    }

    /// Only ever called by whichever `submit` caller found itself the
    /// leader above. Everything between here and the lock actually being
    /// granted (`db::begin_command_batch_leader_tx`'s own `pool.begin()`
    /// plus the real `SELECT ... FOR UPDATE` wait) is exactly the window
    /// other callers keep queueing behind; only once that's done do we
    /// drain the queue (taking everyone who joined, capped at
    /// the configured max batch size), so the batch this call ends up processing can
    /// be considerably larger than one.
    ///
    /// issue #36: this ordering - lock first, drain second - used
    /// to be reversed, which quietly defeated the self-tuning design this
    /// module's own doc comment describes. Draining is unconditional once
    /// the lock step finishes, success or failure alike - a follower that
    /// queued during the wait must never be left behind in a queue nobody
    /// drains again, since leadership for a *new* batch is only ever
    /// assigned to whoever finds the queue empty (see `submit`'s own
    /// `is_leader` check), and a queue nobody drains never goes empty.
    ///
    /// Two further guarantees keep that invariant true:
    /// - **Over-cap remainder**: if more than the configured max batch size commands
    ///   are queued, this leader keeps going - lock, drain, process -
    ///   until a drain empties the queue, answering each batch as it
    ///   commits. No one else can become leader while the queue is
    ///   non-empty, so it must be this one.
    /// - **Cancellation**: `enqueue` runs this on its own task, so a
    ///   leader's caller being dropped (client disconnect, timeout) no
    ///   longer stops it (docs/architecture.md §114). A panic still can:
    ///   a [`LeaderGuard`] stays armed until a drain empties the queue,
    ///   and if this future ends first it fails everything still queued
    ///   (a retryable `BatchFailed`) and empties the queue, so the next
    ///   caller becomes a fresh leader instead of every later caller
    ///   hanging forever.
    #[allow(clippy::too_many_arguments)]
    async fn run_as_leader(
        &self,
        pool: &Pool,
        dispatcher: &dyn CommandDispatcher,
        projection_dispatcher: &dyn ProjectionDispatcher,
        broadcaster: &EventBroadcaster,
        event_cache: &EventCache,
        key_service: Option<&KeyService>,
        bounded_context_name: &str,
        queue: &Queue,
    ) {
        let mut guard = LeaderGuard {
            queue: queue.clone(),
            armed: true,
        };
        let permits = self.leader_permits(pool);
        let mut first_pass = true;

        loop {
            // Held for the whole lock-wait-plus-batch, released at the
            // end of this iteration - see `leader_permits`.
            let _permit = permits
                .acquire()
                .await
                .expect("leader semaphore is never closed");

            // Timed separately from `commit_command_batch`'s own
            // per-phase log - this is the "waiting for the lock" half.
            let lock_wait_started = std::time::Instant::now();
            let leader_tx = crate::db::begin_command_batch_leader_tx(
                pool,
                bounded_context_name,
                Some(self.idle_in_transaction_timeout),
            )
            .await;
            tracing::debug!(
                bounded_context = %bounded_context_name,
                lock_wait_us = lock_wait_started.elapsed().as_micros(),
                "command batch leader lock wait"
            );

            // Never empty on the first pass (the leader's own entry is
            // in there); on later passes only entered when the previous
            // drain reported a remainder, which nobody else can take.
            let (pending, more_remaining) = drain_up_to(queue, self.max_batch_size);
            if !more_remaining {
                // Atomic with the drain above (no await between them):
                // the queue is empty, so leadership is released.
                guard.armed = false;
            }
            let (batch, replies): (
                Vec<BatchedCommand>,
                Vec<oneshot::Sender<BatchedCommandResult>>,
            ) = pending.into_iter().map(|p| (p.command, p.reply)).unzip();

            let outcome = match leader_tx {
                Err(e) => Err(e),
                Ok(leader_tx) => {
                    crate::db::commit_command_batch(
                        leader_tx,
                        dispatcher,
                        projection_dispatcher,
                        key_service,
                        batch,
                        Some(&self.slow),
                    )
                    .await
                }
            };

            match outcome {
                Err(e) => {
                    // Lock acquisition or the shared transaction's final
                    // commit failed - nothing in this batch is
                    // trustworthy, so every command gets an equivalent
                    // error. `Error` isn't `Clone`, so the others get a
                    // `BatchFailed` carrying the original's code and
                    // message; the first command of the first batch -
                    // the one whose caller became leader - keeps the
                    // original typed error, as the leader's own return
                    // value used to.
                    let mut replies = replies.into_iter();
                    let original = if first_pass { replies.next() } else { None };
                    for reply in replies {
                        let _ = reply.send(Err(Error::batch_failed(&e)));
                    }
                    if let Some(reply) = original {
                        let _ = reply.send(Err(e));
                    }
                }
                Ok(results) => {
                    // Every accepted command's own events, across the
                    // whole batch, broadcast together once the shared
                    // commit has actually succeeded.
                    crate::db::broadcast_appended_batch(
                        pool,
                        broadcaster,
                        event_cache,
                        results.iter().flatten(),
                    )
                    .await;
                    for (reply, result) in replies.into_iter().zip(results) {
                        let _ = reply.send(result);
                    }
                }
            }

            first_pass = false;
            if !more_remaining {
                break;
            }
        }
    }
}

/// A queued command's answer. A closed channel means the leader task
/// ended without answering it - only a panic does that, since the task is
/// never aborted.
async fn await_reply(reply: oneshot::Receiver<BatchedCommandResult>) -> BatchedCommandResult {
    reply.await.unwrap_or_else(|_| {
        Err(Error::batch_failed_msg(
            "command batch leader task ended without producing a result",
        ))
    })
}

fn lock_queue(queue: &Queue) -> std::sync::MutexGuard<'_, Vec<PendingCommand>> {
    // Never held across an await and never panics while held, but a
    // poisoned lock must not itself wedge every later caller.
    queue.lock().unwrap_or_else(|e| e.into_inner())
}

/// Takes everything currently queued (up to `max`). The second element
/// is whether entries remain afterwards - if so the queue is *not* empty,
/// so no new leader can arise and the current one must keep draining. When
/// it is `false` the queue is empty and a push that arrives next becomes
/// a fresh leader for a new batch.
fn drain_up_to(queue: &Queue, max: usize) -> (Vec<PendingCommand>, bool) {
    let mut queue = lock_queue(queue);
    if queue.len() <= max {
        (std::mem::take(&mut *queue), false)
    } else {
        (queue.drain(..max).collect(), true)
    }
}

/// Armed for as long as the current leader is the one responsible for
/// eventually emptying the queue. If the leader's future ends while
/// armed (a panic - it runs on its own task, so a caller's disconnect
/// no longer cancels it), everything still queued is failed and the
/// queue emptied so leadership can be re-elected by the next push.
struct LeaderGuard {
    queue: Queue,
    armed: bool,
}

impl Drop for LeaderGuard {
    fn drop(&mut self) {
        if !self.armed {
            return;
        }
        let stranded = std::mem::take(&mut *lock_queue(&self.queue));
        for pending in stranded {
            let _ = pending.reply.send(Err(Error::batch_failed_msg(
                "the command batch leader stopped before processing this command; retry",
            )));
        }
    }
}