Batch Publishing & Partial Failures

publisher.publish_batch::<T>(&msgs) sends many messages in one operation. On the happy path it returns Ok(()) and there is nothing more to think about. The interesting case is the one in the middle: the broker accepted some of the batch and not the rest.

shove reports that case as ShoveError::PartialBatch, which carries the indices into your own msgs slice that still need publishing — so you re-send those records, not the whole batch.

The three outcomes

match publisher.publish_batch::<Orders>(&records).await {
    // Every record confirmed.
    Ok(()) => {}
 
    // Some confirmed, some not. `to_republish()` names the rest.
    Err(ShoveError::PartialBatch(f)) => {
        let retry: Vec<OrderEvent> = f
            .to_republish()
            .iter()
            .filter_map(|&i| records.get(i).cloned())
            .collect();
        publisher.publish_batch::<Orders>(&retry).await?;
    }
 
    // The batch failed as a whole — encoding, topology, an unreachable broker.
    Err(e) => return Err(e),
}

The third arm matters as much as the second. A batch where nothing landed is not partial, so it keeps returning exactly the error it always did: ShoveError::Topology, ShoveError::Serialization, ShoveError::Connection. Existing match arms do not need to change, and code that never looks at PartialBatch behaves as it did before.

What BatchFailure tells you

failed() and unattempted() are disjoint, so both halves of the invariant hold and you can rely on either:

succeeded() + failed().len() + unattempted().len() == msgs.len()
succeeded() + to_republish().len()                 == msgs.len()

ShoveError::is_retryable() on a PartialBatch delegates to source(): re-publishing the outstanding records is worth attempting exactly when the underlying error was.

Index fidelity per backend

Brokers fail differently, and shove reports what each one actually knows rather than flattening it to a lowest common denominator. Both shapes are safe to feed straight into a re-publish; the difference is how tight the set is.

failed() means the broker gave a verdict — nothing else earns it. A transport failure says nothing about whether the record was stored, so RabbitMQ (a failed basic_publish or confirm), NATS (an ack that timed out or died with the connection), and Redis (an XADD reply that never arrived) report those indices as unattempted() instead. The re-publish set is identical either way; only the diagnosis differs.

This is why the payload is a set of indices rather than "the index of the first failure". On the sparse backends a record after a failure has usually already succeeded, so "re-publish everything from the first failure onwards" re-sends records that were fine. Its inverse — "skip the first succeeded records" — is worse: it loses data, because succeeded is a count, not a prefix length.

Duplicates over loss

A record whose fate is genuinely unknown — submitted to the broker but never confirmed — is reported as unattempted, never as succeeded. RabbitMQ hits this when a basic_publish fails partway through a batch: earlier messages are on the channel, but nothing has confirmed them, and the client cannot tell whether the broker received them.

shove always resolves that ambiguity toward re-publishing. So make your consumers idempotent — that is the one thing a to_republish() retry loop asks of you.

Deduplicate on a key from your own payload, not on shove's message id. The x-message-id / Shove-Message-Id / Nats-Msg-Id header each backend stamps is minted fresh on every publish call, so a re-published record arrives with a different id — those headers deduplicate a broker-side redelivery of one publish, not a deliberate re-publish of the same record. The one exception is SNS FIFO, whose MessageDeduplicationId is derived from the payload content and so does collapse an identical re-publish inside its dedup window.

See Exactly-Once (RabbitMQ) for the strongest available guarantee.

Upgrading from 0.12

The generic Publisher<B>::publish_batch has always returned Result<()> and is unchanged. The six concrete publishers — RabbitMqPublisher, KafkaPublisher, NatsPublisher, RedisPublisher, SnsPublisher, InMemoryPublisher — used to return (u64, Result<()>), where the u64 was a count of confirmed records. They now return Result<()> like the generic one.

// 0.12
let (confirmed, result) = publisher.publish_batch::<Orders>(&records).await;
 
// 0.14
let result = publisher.publish_batch::<Orders>(&records).await;

The count is not lost, and it is no longer a bare number you have to interpret:

match publisher.publish_batch::<Orders>(&records).await {
    Ok(()) => { /* confirmed == records.len() */ }
    Err(ShoveError::PartialBatch(f)) => {
        let confirmed = f.succeeded();
        let outstanding = f.to_republish();
        // ...
    }
    Err(_) => { /* confirmed == 0 */ }
}

A count alone could never say which records were confirmed, which is the question a retry actually has to answer. On the sparse backends it was actively misleading: succeeded is a count, not a prefix length, so "skip the first succeeded records" loses data.

Metrics

publish_batch splits shove_messages_published_total by what the backend confirmed: outcome="success" for succeeded() records and outcome="error" for the rest. A partial batch therefore increments both, and the two always sum to the batch size. shove_message_publish_duration_seconds records one sample for the whole call — the user-observable latency — regardless of how many messages were inside. See Observability for the full schema.