Plan: outbox-drainer lease-based three-phase refactor

22:54:50.491  pid=92  canopy_appeals    LWLock:WALWrite     COMMIT
22:54:50.491  pid=79  canopy_security   LWLock:WALWrite     COMMIT
22:54:50.491  pid=638 canopy_appeals    IO:WalSync          COMMIT
22:54:50.491  pid=638 canopy_appeals    idle in transaction UPDATE event_outbox SET published_at = now() WHERE id = $1
22:54:50.763  pid=676 canopy_persons    idle in transaction UPDATE event_outbox SET published_at = now() WHERE id = $1
22:54:51.036  pid=674 canopy_applications idle in transaction UPDATE event_outbox SET published_at = now() WHERE id = $1
22:54:51.309  pid=1066 canopy_security   IO:WalSync          COMMIT  (>800ms)

async fn drain_once(
    pool: &PgPool,
    manager: &ConnectionManager,
    cfg: &DrainerConfig,   // batch_size, lease_ttl_secs, pipeline_depth, drainer_id
) -> Result<DrainStats, DrainError> {
    // ----- Phase 1: claim (one statement, short tx) -----
    // CTE form is required: Postgres disallows FOR UPDATE inside a
    // subquery used in `WHERE id IN (SELECT …)`. The CTE form is the
    // canonical skip-locked-claim pattern.
    let claimed: Vec<(Uuid, JsonValue, i32)> = sqlx::query_as(
        "WITH locked AS (
            SELECT id FROM event_outbox
             WHERE published_at IS NULL
               AND (claimed_at IS NULL
                    OR claimed_at < now() - ($2::bigint * interval '1 second'))
             ORDER BY enqueued_at
             FOR UPDATE SKIP LOCKED
             LIMIT $3
         )
         UPDATE event_outbox
            SET claimed_at = now(), claimed_by = $1
           FROM locked
          WHERE event_outbox.id = locked.id
         RETURNING event_outbox.id, event_outbox.payload, event_outbox.attempts"
    )
    .bind(&cfg.drainer_id)
    .bind(cfg.lease_ttl_secs as i64)
    .bind(cfg.batch_size)
    .fetch_all(pool).await?;

    if claimed.is_empty() { return Ok(DrainStats::empty()); }

    // ----- Phase 2: publish outside any tx, channel-per-batch -----
    //
    // Any early-return path here must release the active claims so the
    // next drain tick can retry, instead of stranding 100 rows for the
    // entire lease TTL. The publish phase is wrapped in an inner closure
    // returning Result<(confirmed, failed), DrainError>; on outer Err we
    // call `release_claims_only(pool, &all_claimed_ids, drainer_id).await`
    // before propagating. `release_claims_only` is a single bulk UPDATE
    // that clears `claimed_at`/`claimed_by` without bumping `attempts`
    // or touching `last_error` — the failure was infrastructural (channel
    // open, confirm_select), not a per-message problem.
    //
    // We deliberately do NOT set `mandatory=true`. In a fan-out topic
    // event bus, "no queue currently bound for this routing key" is
    // valid during dev/test or while a subscribing service is down.
    // mandatory=true would mark such events as failed and retry them
    // forever; instead we treat broker-acceptance as published.
    // Consumers with already-declared durable queues + bindings receive
    // events that arrive after their bindings exist; events published
    // before any binding existed for that routing key are dropped by
    // the broker. This is the accepted trade-off — the durable outbox
    // guarantees event durability at-the-producer, not subscriber-side
    // backfill. Broker-side delivery is the broker's job, not the
    // drainer's.
    let publish_result: Result<(Vec<Uuid>, Vec<(Uuid, String)>), DrainError> = async {
        let channel = manager.current_channel().await?;
        channel.confirm_select(ConfirmSelectOptions::default()).await?;

        let mut confirmed_ids = Vec::new();
        let mut failed: Vec<(Uuid, String)> = Vec::new();

        for chunk in claimed.chunks(cfg.pipeline_depth) {
            let mut pending = Vec::with_capacity(chunk.len());
            for (id, payload, _attempts) in chunk {
                let env = match serde_json::from_value::<EventEnvelope>(payload.clone()) {
                    Err(e) => { failed.push((*id, format!("deserialise: {e}"))); continue; }
                    Ok(env) => env,
                };
                let bytes = match serde_json::to_vec(&env) {
                    Err(e) => { failed.push((*id, format!("serialise: {e}"))); continue; }
                    Ok(b) => b,
                };
                // basic_publish returning an error is a per-message
                // failure (channel/broker hiccup for this publish);
                // record it and proceed, do not bail.
                let confirm = match channel.basic_publish(
                    EVENTS_EXCHANGE.into(),
                    env.event_type.as_str().into(),
                    BasicPublishOptions::default(),
                    &bytes,
                    BasicProperties::default()
                        .with_content_type("application/json".into())
                        .with_delivery_mode(2),
                ).await {
                    Err(e) => { failed.push((*id, format!("publish: {e}"))); continue; }
                    Ok(c) => c,
                };
                pending.push((*id, confirm));
            }
            for (id, confirm) in pending {
                // Lapin's Confirmation enum:
                //   Ack(None)                       — broker accepted, normal case → mark published
                //   Ack(Some(BasicReturnMessage))   — only seen with mandatory=true (unroutable); we
                //                                     don't set mandatory, so this should not occur,
                //                                     but be defensive: treat as failed publish
                //   Nack(_)                         — broker rejected (e.g. resource limit reached)
                //                                     → bump attempts, retry next tick
                match confirm.await {
                    Ok(Confirmation::Ack(None)) => confirmed_ids.push(id),
                    Ok(Confirmation::Ack(Some(ret))) => {
                        failed.push((id, format!("ack with return: {ret:?}")));
                    }
                    Ok(Confirmation::Nack(reason)) => {
                        failed.push((id, format!("nack: {reason:?}")));
                    }
                    Ok(Confirmation::NotRequested) => unreachable!("confirm_select was called"),
                    Err(e) => failed.push((id, format!("amqp: {e}"))),
                }
            }
        }
        Ok((confirmed_ids, failed))
    }.await;

    let (confirmed_ids, failed) = match publish_result {
        Ok(pair) => pair,
        Err(e) => {
            // Infrastructure failure (channel open, confirm_select). Don't
            // bump attempts (nothing was wrong with the messages). Just
            // release every claim we took in Phase 1.
            let all_ids: Vec<Uuid> = claimed.iter().map(|(id, _, _)| *id).collect();
            release_claims_only(pool, &all_ids, &cfg.drainer_id).await?;
            return Err(e);
        }
    };

    // ----- Phase 3: mark results in two short txes, guarded by claimed_by -----
    if !confirmed_ids.is_empty() {
        sqlx::query(
            "UPDATE event_outbox
                SET published_at = now(), claimed_at = NULL, claimed_by = NULL
              WHERE id = ANY($1) AND claimed_by = $2"
        )
        .bind(&confirmed_ids).bind(&cfg.drainer_id)
        .execute(pool).await?;
    }
    if !failed.is_empty() {
        // Bulk failure update: bump attempts, store last_error, clear claim.
        // unnest carries the per-row error string in lockstep with the id.
        let (ids, errs): (Vec<_>, Vec<_>) = failed.iter().cloned().unzip();
        sqlx::query(
            "UPDATE event_outbox AS o
                SET attempts = o.attempts + 1,
                    last_error = f.err,
                    claimed_at = NULL,
                    claimed_by = NULL
              FROM unnest($1::uuid[], $2::text[]) AS f(id, err)
              WHERE o.id = f.id AND o.claimed_by = $3"
        )
        .bind(&ids).bind(&errs).bind(&cfg.drainer_id)
        .execute(pool).await?;
    }

    Ok(DrainStats {
        claimed: claimed.len(),
        confirmed: confirmed_ids.len(),
        failed: failed.len(),
    })
}

/// Bulk-clear claims without bumping attempts. Used when Phase 2
/// hits an infrastructure error (channel/confirm_select) before any
/// per-message work — the messages are blameless, the drainer hit
/// transient broker trouble. Releasing the claim lets the next tick
/// retry the batch with a fresh channel.
async fn release_claims_only(
    pool: &PgPool,
    ids: &[Uuid],
    drainer_id: &str,
) -> Result<(), sqlx::Error> {
    if ids.is_empty() { return Ok(()); }
    sqlx::query(
        "UPDATE event_outbox
            SET claimed_at = NULL, claimed_by = NULL
          WHERE id = ANY($1) AND claimed_by = $2"
    )
    .bind(ids).bind(drainer_id)
    .execute(pool).await?;
    Ok(())
}

CREATE INDEX event_outbox_lease_idx
    ON event_outbox (claimed_at NULLS FIRST, enqueued_at)
    WHERE published_at IS NULL;

assert!(cfg.batch_size > 0, "CANOPY_MQ_DRAINER_BATCH_SIZE must be > 0; got 0 would make claimed.chunks(0) panic");
assert!(cfg.pipeline_depth > 0, "CANOPY_MQ_DRAINER_PIPELINE_DEPTH must be > 0; got 0 would make claimed.chunks(0) panic");

// Heuristic: assume a worst-case per-publish latency of 100ms (broker
// stress + TLS handshake). pipeline_depth × 100ms must stay well under
// lease_ttl, with ≥3× safety margin so a slow batch can't expire its
// own lease mid-publish.
let assumed_max_publish_ms = 100u64;
let worst_case_batch_ms = (cfg.pipeline_depth as u64) * assumed_max_publish_ms;
let lease_ttl_ms = cfg.lease_ttl_secs * 1000;
assert!(
    lease_ttl_ms >= worst_case_batch_ms * 3,
    "CANOPY_MQ_DRAINER_LEASE_TTL_SECS ({lease}s) must be ≥ 3× worst-case batch publish time \
     (pipeline_depth × 100ms = {worst}ms). Either raise lease_ttl_secs or lower pipeline_depth.",
    lease = cfg.lease_ttl_secs,
    worst = worst_case_batch_ms,
);

-- SPDX-License-Identifier: AGPL-3.0-or-later
-- Outbox-drainer lease columns + lease-aware partial index.
-- See: docs/modules/ROOT/pages/plans/archive/outbox-drainer-lease-refactor.adoc

ALTER TABLE event_outbox
    ADD COLUMN claimed_at TIMESTAMPTZ NULL,
    ADD COLUMN claimed_by TEXT NULL;

CREATE INDEX event_outbox_lease_idx
    ON event_outbox (claimed_at NULLS FIRST, enqueued_at)
    WHERE published_at IS NULL;

UPDATE event_outbox AS o
   SET attempts = o.attempts + 1,
       last_error = f.err,
       claimed_at = NULL,
       claimed_by = NULL
  FROM unnest($1::uuid[], $2::text[]) AS f(id, err)
 WHERE o.id = f.id AND o.claimed_by = $3