@@ -147,7 +147,8 @@ docstrings everywhere. README stays short and links to the docs.
 
 * Calling into Python from Rust threads (events are pulled; see ADR 0004).
 * Threads or Tokio runtimes outside `crates/fanferro-py/src/runtime.rs`, or awaitables not created
-  with `runtime::future_into_py`: they would outlive the `atexit` shutdown of the runtime.
+  with `runtime::future_into_py` (or `runtime::Completions` for settlements): they would outlive the
+  `atexit` shutdown of the runtime.
 * Holding a `parking_lot` lock across `.await`.
 * Long `wait_time` in tests; `time.sleep` in async code; global state in Python modules.
 * Logging message bodies (they may contain personal data) - log ids and types.
 
@@ -28,9 +28,18 @@ Without any tuning, clients and consumers now perform like a hand-tuned 0.1 conf
 - **Slots are freed when settlement starts**: a handler that returns hands its slot to the next
   message while the delete is in flight; settlements in progress are bounded separately.
 - **Batched acks**: deletes and visibility changes wait up to `settle_linger` (5 ms) to fill a
-  batch of ten, and are sent at once when every running handler awaits an explicit settlement.
+  batch of ten, and are sent at once when every running handler awaits an explicit settlement
+  in that batch.
+- **Automatic acks are not awaited per message**: the engine deletes the message in the
+  background (`consumer.stop()` waits for it). `consumer.stats()["processed"]` and the
+  `message.processed` event count handlers that succeeded; `stats()["engine"]["messages_acked"]`
+  counts confirmed deletes, and a failed delete is reported as a `message.ack_failed` event. Explicit
+  settlements (`await message.ack()`...) still resolve once SQS confirmed them.
 - `consumer.stop()` and `client.close()` no longer wait up to half a second for the dispatcher or
   the event pump, and events emitted just before `close()` are no longer lost.
+- Per-message work in the consumer and publisher is lighter: settlement awaitables are resolved
+  in batches, event checks, type names and encoders are cached, and messages are no longer copied
+  while being decoded.
 
 Migrating from 0.1:
 
 
@@ -67,8 +67,8 @@ impl Default for HeartbeatConfig {
 #[serde(default, deny_unknown_fields)]
 pub struct SettleConfig {
     /// Seconds a batch waits for more settlements (at most 1). A batch is sent as soon as it
-    /// holds ten deletes or ten visibility changes, or when a settlement is urgent; 0 sends
-    /// whatever is queued immediately.
+    /// holds ten deletes or ten visibility changes, or when no other settlement can join it;
+    /// 0 sends whatever is queued immediately.
     pub linger: f64,
     /// Retries for failed delete/visibility calls.
     pub retry: RetrySettings,
 
@@ -10,18 +10,18 @@
 
 use std::collections::BTreeMap;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::time::Duration;
 
 use parking_lot::Mutex;
 use time::OffsetDateTime;
 use time::format_description::well_known::Rfc3339;
-use tokio::sync::{OwnedSemaphorePermit, oneshot};
+use tokio::sync::{OwnedSemaphorePermit, mpsc, oneshot};
 use tokio::time::Instant;
 
 use super::EngineInner;
 use super::config::MAX_VISIBILITY;
-use super::settle::{SettleCommand, SettleKind};
+use super::settle::{Reply, SettleCommand, SettleKind};
 use crate::backend::{ReceivedMessage, SendRequest};
 use crate::error::{AwsErrorKind, Error, Result};
 use crate::events::{Event, EventKind};
@@ -92,8 +92,10 @@ pub struct Lease {
     /// Settling slot: held while the settlement is in progress.
     settling: Mutex<Option<OwnedSemaphorePermit>>,
     settled: AtomicBool,
-    /// The next settlement command is urgent (see [`Lease::expedite`]).
-    expedite: AtomicBool,
+    /// The next settlement is awaited by a handler (see [`Lease::mark_awaited`]).
+    awaited: AtomicBool,
+    /// How many handlers wait, when all of them do (0: unknown; see [`Lease::mark_awaited`]).
+    waiting_handlers: AtomicUsize,
 }
 
 impl std::fmt::Debug for Lease {
@@ -129,7 +131,8 @@ impl Lease {
             permit: Mutex::new(Some(permit)),
             settling: Mutex::new(None),
             settled: AtomicBool::new(false),
-            expedite: AtomicBool::new(false),
+            awaited: AtomicBool::new(false),
+            waiting_handlers: AtomicUsize::new(0),
         }
     }
 
@@ -170,19 +173,35 @@ impl Lease {
     /// meanwhile), then hands the in-flight slot back so that receiving continues while the
     /// settlement completes. The lease stays tracked until [`Lease::finish`].
     async fn begin(&self) -> Result<()> {
+        self.claim()?;
+        self.start_settling().await;
+        Ok(())
+    }
+
+    /// Marks the lease as being settled; fails if a settlement was requested already.
+    pub(crate) fn claim(&self) -> Result<()> {
         if self.settled.swap(true, Ordering::SeqCst) {
             return Err(Error::Lease(format!("message {} was already settled", self.message.message_id)));
         }
+        Ok(())
+    }
+
+    pub(crate) async fn start_settling(&self) {
         // The semaphore is never closed, so this only fails if the engine is gone.
-        if let Ok(permit) = self.inner.settling.clone().acquire_owned().await {
+        let permit = self.inner.settling.clone().acquire_owned().await.ok();
+        self.enter_settling(permit);
+    }
+
+    /// Takes the settling slot, then hands the in-flight slot back.
+    fn enter_settling(&self, permit: Option<OwnedSemaphorePermit>) {
+        if let Some(permit) = permit {
             *self.settling.lock() = Some(permit);
             self.inner.metrics.settling.add(1);
         }
         if let Some(entry) = self.inner.leases.lock().get_mut(&self.id) {
             entry.status = LeaseStatus::Settling;
         }
         self.permit.lock().take();
-        Ok(())
     }
 
     /// Returns the settling slot (if one is held).
@@ -229,19 +248,38 @@ impl Lease {
     }
 
     async fn settle(&self, kind: SettleKind) -> Result<()> {
-        let sender = self.inner.settle_sender().ok_or_else(|| Error::Shutdown("the consumer is closed".into()))?;
         let (reply, receiver) = oneshot::channel();
-        let mut command = SettleCommand::new(self.message.receipt_handle.clone(), kind, Some(reply));
-        command.urgent = self.expedite.swap(false, Ordering::SeqCst);
-        sender.send(command).map_err(|_| Error::Shutdown("the consumer is closed".into()))?;
+        self.queue(kind, Reply::Channel(reply));
         receiver.await.map_err(|_| Error::Shutdown("the consumer closed before settling".into()))?
     }
 
-    /// Sends this lease's next settlement (or explicit extension) right away instead of letting
-    /// it wait to fill a batch. Use it when the caller is blocked on that settlement and no
-    /// other settlement is expected soon (e.g. every handler awaits an explicit ack).
-    pub fn expedite(&self) {
-        self.expedite.store(true, Ordering::SeqCst);
+    /// Hands a settlement command to the settle loop, which sends the outcome to `reply` (right
+    /// away if the consumer is closed).
+    pub(crate) fn queue(&self, kind: SettleKind, reply: Reply) {
+        let Some(sender) = self.inner.settle_sender() else {
+            reply.send(Err(Error::Shutdown("the consumer is closed".into())));
+            return;
+        };
+        let mut command = SettleCommand::new(self.message.receipt_handle.clone(), kind, Some(reply));
+        command.awaited = self.awaited.swap(false, Ordering::SeqCst);
+        command.waiting_handlers = Some(self.waiting_handlers.swap(0, Ordering::SeqCst)).filter(|&n| n > 0);
+        if let Err(mpsc::error::SendError(command)) = sender.send(command)
+            && let Some(reply) = command.reply
+        {
+            reply.send(Err(Error::Shutdown("the consumer is closed".into())));
+        }
+    }
+
+    /// Notes that a handler waits for this lease's next settlement (or explicit extension), so
+    /// that the batch it joins is not held back for settlements that cannot come.
+    ///
+    /// `waiting_handlers`: when every handler that could request another settlement waits for
+    /// one, how many handlers wait. The settle loop sends the batch as soon as it holds that many
+    /// awaited settlements; while fewer are pending, some handlers wait for settlements that were
+    /// sent already, and their slots will soon go to handlers that settle more.
+    pub fn mark_awaited(&self, waiting_handlers: Option<usize>) {
+        self.awaited.store(true, Ordering::SeqCst);
+        self.waiting_handlers.store(waiting_handlers.unwrap_or(0), Ordering::SeqCst);
     }
 
     fn cap_visibility(&self, timeout: Duration) -> Duration {
@@ -252,12 +290,35 @@ impl Lease {
 
     /// Deletes the message (successful processing).
     pub async fn ack(&self) -> Result<()> {
-        self.begin().await?;
+        self.claim()?;
+        self.ack_claimed().await
+    }
+
+    /// [`Lease::ack`] after [`Lease::claim`] succeeded.
+    pub(crate) async fn ack_claimed(&self) -> Result<()> {
+        self.start_settling().await;
         let result = self.settle(SettleKind::Delete).await;
         self.finish(Outcome::Acked, &result);
         result
     }
 
+    /// The fast path of an ack after [`Lease::claim`]: when a settling slot is free (the usual
+    /// case), takes it and queues the delete with `reply` right away, so acks requested one after
+    /// another reach the settle loop in that order. Hands `reply` back when no slot is free.
+    pub(crate) fn try_queue_ack(&self, reply: Reply) -> std::result::Result<(), Reply> {
+        let Ok(permit) = self.inner.settling.clone().try_acquire_owned() else {
+            return Err(reply);
+        };
+        self.enter_settling(Some(permit));
+        self.queue(SettleKind::Delete, reply);
+        Ok(())
+    }
+
+    /// Records the outcome of an ack whose delete was queued with [`Lease::queue`].
+    pub(crate) fn finish_ack(&self, result: &Result<()>) {
+        self.finish(Outcome::Acked, result);
+    }
+
     /// Makes the message visible again after `delay` (retry with backoff).
     pub async fn nack(&self, delay: Duration) -> Result<()> {
         self.begin().await?;
 
@@ -27,7 +27,9 @@ mod lease;
 mod poller;
 mod settle;
 
+use std::borrow::Cow;
 use std::collections::{BTreeMap, HashMap, VecDeque};
+use std::future::Future;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Duration;
@@ -55,7 +57,7 @@ use crate::types::attributes::{md5_of_attributes, md5_of_body};
 use crate::types::reserved;
 use config::Settings;
 use lease::LeaseEntry;
-use settle::{SettleCommand, SettleKind};
+use settle::{Reply, SettleCommand, SettleKind};
 
 /// A received message handed to the application.
 #[derive(Debug)]
@@ -75,6 +77,73 @@ impl Delivery {
     pub fn message(&self) -> &ReceivedMessage {
         self.lease.message()
     }
+
+    /// Acks the message without waiting for the delete. The settlement runs on the engine's
+    /// runtime and is tracked ([`ConsumerEngine::close`] waits for it); its outcome is reported
+    /// through metrics (`messages_acked`, `ack_failures`) and events (`message.acked`,
+    /// `message.ack_failed`).
+    ///
+    /// # Errors
+    ///
+    /// [`Error::Lease`] if the message was settled (or its settlement started) already.
+    pub fn ack_in_background(self: &Arc<Self>) -> Result<()> {
+        self.lease.claim()?;
+        // Failures are reported by the lease (metrics and events).
+        self.continue_ack(|_| {});
+        Ok(())
+    }
+
+    /// Acks the message and hands the outcome to `done`. The delete is queued before this
+    /// returns (when a settling slot is free), so acks requested one after another reach the
+    /// settle loop in that order; `done` runs once the delete completed.
+    pub fn spawn_ack<Done>(self: &Arc<Self>, done: Done)
+    where
+        Done: FnOnce(Result<()>) + Send + 'static,
+    {
+        match self.lease.claim() {
+            Ok(()) => self.continue_ack(done),
+            Err(error) => done(Err(error)),
+        }
+    }
+
+    /// An ack after its lease was claimed. Usually the delete is queued right away and finished by
+    /// the settle loop (no task per ack); without a free settling slot, a tracked task waits for
+    /// one.
+    fn continue_ack<Done>(self: &Arc<Self>, done: Done)
+    where
+        Done: FnOnce(Result<()>) + Send + 'static,
+    {
+        let delivery = self.clone();
+        let finish = Reply::Callback(Box::new(move |result: Result<()>| {
+            delivery.lease.finish_ack(&result);
+            done(result);
+        }));
+        if let Err(finish) = self.lease.try_queue_ack(finish) {
+            let delivery = self.clone();
+            let inner = &self.lease.inner;
+            inner.settlements.spawn_on(
+                async move {
+                    delivery.lease.start_settling().await;
+                    delivery.lease.queue(SettleKind::Delete, finish);
+                },
+                &inner.runtime,
+            );
+        }
+    }
+
+    /// Runs `settle` (an ack, retry, release... of this delivery) on the engine's runtime and
+    /// hands its outcome to `done`. Like background acks, the settlement is tracked:
+    /// [`ConsumerEngine::close`] waits for it, and it completes even if nobody waits for `done`.
+    pub fn spawn_settlement<Settle, Fut, Done>(self: &Arc<Self>, settle: Settle, done: Done)
+    where
+        Settle: FnOnce(Arc<Self>) -> Fut + Send + 'static,
+        Fut: Future<Output = Result<()>> + Send + 'static,
+        Done: FnOnce(Result<()>) + Send + 'static,
+    {
+        let delivery = self.clone();
+        let inner = &self.lease.inner;
+        inner.settlements.spawn_on(async move { done(settle(delivery).await) }, &inner.runtime);
+    }
 }
 
 /// Lifecycle state of the engine.
@@ -122,6 +191,8 @@ pub(crate) struct EngineInner {
     pollers: Mutex<Vec<JoinHandle<()>>>,
     background: Mutex<Vec<JoinHandle<()>>>,
     decoders: TaskTracker,
+    /// Settlements running in the background (see [`Delivery::ack_in_background`]).
+    settlements: TaskTracker,
     rate_limiter: Option<RateLimiter>,
     runtime: tokio::runtime::Handle,
     /// Until when an aborted long poll may still be served by the queue (see
@@ -231,7 +302,7 @@ impl EngineInner {
 
     async fn decode(self: &Arc<Self>, lease: Lease) -> Option<Delivery> {
         let received_at = OffsetDateTime::now_utc();
-        let message = lease.message().clone();
+        let message = lease.message.clone();
         if self.settings.verify_md5 {
             let body_ok =
                 message.md5_of_body.as_ref().is_none_or(|md5| md5.eq_ignore_ascii_case(&md5_of_body(&message.body)));
@@ -256,9 +327,9 @@ impl EngineInner {
                     self.handle_sns_control(lease, unwrapped.info).await;
                     return None;
                 }
-                (unwrapped.message, unwrapped.attributes, Some(unwrapped.info))
+                (Cow::Owned(unwrapped.message), unwrapped.attributes, Some(unwrapped.info))
             }
-            Ok(None) => (message.body.clone(), message.attributes.clone(), None),
+            Ok(None) => (Cow::Borrowed(message.body.as_str()), message.attributes.clone(), None),
             Err(error) => {
                 self.metrics.decode_failures.inc();
                 return Some(Delivery { lease, sns: None, decoded: Err(error), received_at });
@@ -418,6 +489,7 @@ impl ConsumerEngine {
                 pollers: Mutex::new(Vec::new()),
                 background: Mutex::new(Vec::new()),
                 decoders: TaskTracker::new(),
+                settlements: TaskTracker::new(),
                 rate_limiter,
                 runtime,
                 orphaned_receives_until: Mutex::new(None),
@@ -576,6 +648,9 @@ impl ConsumerEngine {
     /// and after in-flight handlers finished or were cancelled.
     pub async fn close(&self, release_unsettled: bool) {
         self.stop_receiving(false).await;
+        // Background acks complete first: they need the settle loop, which stops below.
+        self.inner.settlements.close();
+        self.inner.settlements.wait().await;
         for delivery in self.drain_buffered() {
             if release_unsettled {
                 let _ = delivery.lease.release().await;
@@ -591,7 +666,7 @@ impl ConsumerEngine {
                     let mut command = SettleCommand::new(
                         entry.receipt_handle,
                         SettleKind::Visibility(self.inner.visibility_floor(Duration::ZERO)),
-                        Some(reply),
+                        Some(Reply::Channel(reply)),
                     );
                     command.urgent = true;
                     if sender.send(command).is_ok() {