aion_server/worker/

envelope.rs

//! Activity task-envelope generation and completion fencing.

use std::collections::hash_map::Entry;
use std::collections::{HashMap, HashSet};
use std::sync::{Arc, Mutex, MutexGuard};

use aion_core::{ActivityId, RunId, WorkflowId};
use sha2::{Digest, Sha256};
use uuid::Uuid;

use crate::error::{CompletionRejectionReason, ServerError};

type ExecutionKey = (WorkflowId, ActivityId);

/// Opaque proof that a worker owns one dispatched execution generation.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct CompletionToken(String);

impl CompletionToken {
    /// Parse a worker-echoed token without assigning meaning to its contents.
    ///
    /// # Errors
    ///
    /// Returns a typed compatibility refusal when a pre-fencing worker omits it.
    pub fn from_wire(
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
        value: String,
    ) -> Result<Self, ServerError> {
        if value.is_empty() {
            return Err(rejection(
                workflow_id,
                activity_id,
                CompletionRejectionReason::MissingCompletionToken,
            ));
        }
        Ok(Self(value))
    }

    /// Return the opaque wire representation.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }

    /// Build a non-wire token for crate-local unit fixtures.
    #[cfg(test)]
    #[must_use]
    pub(crate) fn for_test() -> Self {
        Self("test-generation".to_owned())
    }
}

/// Every completion token authorized for ONE attempt of ONE execution
/// generation of one execution site.
///
/// The vector holds more than one token in exactly one situation: the SAME
/// attempt was delivered more than once, so two live workers each genuinely
/// hold a token minted for the same work. Delivery is at-least-once, so that
/// situation is ordinary rather than exceptional. Both tokens are acceptable
/// candidates; the FIRST accepted completion consumes the whole generation, so
/// the second worker's completion is the duplicate and is refused.
#[derive(Clone, Debug, Eq, PartialEq)]
struct SiteGeneration {
    /// Run-scoped external-effect key of the generation these tokens belong to,
    /// and the execution-generation discriminator. Not a second discriminator
    /// invented here: it is the same [`idempotency_key`] the wire already
    /// carries to the worker, so a reset or continue-as-new — which mints a new
    /// [`RunId`] — is a different generation by construction.
    idempotency_key: String,
    /// One-based delivery attempt these tokens were issued for. A higher
    /// attempt supersedes; an equal attempt is a redelivery of the same work.
    attempt: u32,
    /// Tokens issued for `attempt` that have been neither consumed by an
    /// accepted completion nor revoked by the dispatch that issued them.
    outstanding: Vec<CompletionToken>,
}

/// The generation one accepted completion consumed, handed back so a settlement
/// that fails afterwards can restore exactly what was taken.
///
/// Restoring needs the `idempotency_key` and `attempt` that were consumed;
/// re-deriving either at the restore site would be inventing a second
/// execution-generation discriminator, so the consumed generation travels out
/// with the acceptance instead.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct AcceptedGeneration(SiteGeneration);

impl AcceptedGeneration {
    /// One-based delivery attempt whose completion was accepted.
    #[must_use]
    pub const fn attempt(&self) -> u32 {
        self.0.attempt
    }
}

/// Process-incarnation registry of the only generations allowed to complete.
///
/// Recovery deliberately starts with an empty registry. Re-dispatch issues a
/// fresh token, so a result carrying any pre-recovery token is refused whether
/// it arrives before or after the new dispatch.
#[derive(Clone, Debug, Default)]
pub struct CompletionFences {
    current: Arc<Mutex<HashMap<ExecutionKey, SiteGeneration>>>,
    /// The lease gate (WA-010 R3): tokens whose `ActivityLeased` append is in
    /// flight. A completion for a token that is still here waits at
    /// [`Self::lease_settled`] before the fence accepts it, so a worker that
    /// answers faster than the lease lands cannot put its terminal ahead of
    /// its lease in the history. Armed per delivery attempt and settled by
    /// the handoff either way, so nothing waits on a lease that will never be
    /// recorded.
    lease_gate: Arc<LeaseGate>,
}

#[derive(Debug, Default)]
struct LeaseGate {
    pending: Mutex<HashSet<String>>,
    settled: tokio::sync::Notify,
}

impl CompletionFences {
    /// Authorize one more delivery of one attempt and return its token.
    ///
    /// One generation is held per execution site, and what a call is decides
    /// what happens to it:
    ///
    /// - a different `idempotency_key` is a different EXECUTION GENERATION
    ///   (reset / continue-as-new): the prior generation is superseded whole;
    /// - a higher `attempt` within the same generation is a genuine RETRY: the
    ///   prior attempt's tokens are superseded;
    /// - the SAME attempt within the same generation is a REDELIVERY of work a
    ///   worker may still be executing, so the new token JOINS the outstanding
    ///   set rather than replacing it. Both workers hold a token that can be
    ///   accepted, and the first accepted completion consumes both.
    ///
    /// A re-dispatch of an attempt that has already been superseded registers
    /// nothing: it is logged with both attempt numbers, and the token it returns
    /// is refused as a stale generation when presented, which is the truthful
    /// answer for a worker whose attempt no longer exists.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] when fence state cannot be trusted.
    pub fn issue(
        &self,
        workflow_id: &WorkflowId,
        run_id: &RunId,
        activity_id: &ActivityId,
        attempt: u32,
    ) -> Result<CompletionToken, ServerError> {
        let token = CompletionToken(Uuid::new_v4().to_string());
        let issued_key = idempotency_key(workflow_id, run_id, activity_id);
        let mut state = self.state()?;
        match state.entry((workflow_id.clone(), activity_id.clone())) {
            Entry::Vacant(slot) => {
                slot.insert(SiteGeneration {
                    idempotency_key: issued_key,
                    attempt,
                    outstanding: vec![token.clone()],
                });
            }
            Entry::Occupied(mut slot) => {
                let current = slot.get_mut();
                if current.idempotency_key != issued_key || attempt > current.attempt {
                    *current = SiteGeneration {
                        idempotency_key: issued_key,
                        attempt,
                        outstanding: vec![token.clone()],
                    };
                } else if attempt == current.attempt {
                    current.outstanding.push(token.clone());
                } else {
                    tracing::warn!(
                        workflow_id = %workflow_id,
                        activity_id = %activity_id,
                        superseded_attempt = attempt,
                        current_attempt = current.attempt,
                        "re-dispatch of an already superseded attempt registers no generation; \
                         its completion will be refused as a stale generation"
                    );
                }
            }
        }
        Ok(token)
    }

    /// Consume the current generation when `submitted` is one of the tokens it
    /// still has outstanding.
    ///
    /// Comparison and consumption share one mutex critical section, so two
    /// concurrent submissions cannot both become truth: the whole generation is
    /// removed by the first, and every later presentation of any of its tokens
    /// finds no generation at all.
    ///
    /// # Errors
    ///
    /// Returns a typed rejection for a missing generation or a token that is
    /// not outstanding in it, or [`ServerError::LockPoisoned`] when fence state
    /// cannot be trusted.
    pub fn accept(
        &self,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
        submitted: &CompletionToken,
    ) -> Result<AcceptedGeneration, ServerError> {
        self.wait_lease_settled_blocking(submitted)?;
        let mut state = self.state()?;
        let Entry::Occupied(slot) = state.entry((workflow_id.clone(), activity_id.clone())) else {
            return Err(rejection(
                workflow_id,
                activity_id,
                CompletionRejectionReason::NoCurrentGeneration,
            ));
        };
        if !slot.get().outstanding.contains(submitted) {
            return Err(rejection(
                workflow_id,
                activity_id,
                CompletionRejectionReason::StaleGeneration,
            ));
        }
        Ok(AcceptedGeneration(slot.remove()))
    }

    /// Restore a consumed generation after accepted-path settlement fails.
    ///
    /// A concurrently issued newer generation always wins; the accepted
    /// generation is restored only while the execution site has no current one.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] when fence state cannot be trusted.
    pub fn restore_if_absent(
        &self,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
        accepted: &AcceptedGeneration,
    ) -> Result<(), ServerError> {
        self.state()?
            .entry((workflow_id.clone(), activity_id.clone()))
            .or_insert_with(|| accepted.0.clone());
        Ok(())
    }

    /// Revoke exactly the token this caller issued, and nothing else.
    ///
    /// A dispatch that could not place its task withdraws its OWN authorization.
    /// It never withdraws a sibling token still held by a live worker executing
    /// the same attempt, and it never disturbs a newer retry — a newer attempt
    /// already replaced the outstanding set, so the old token is simply absent
    /// and the removal is a no-op. The generation is dropped once its last
    /// outstanding token is gone.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] when fence state cannot be trusted.
    pub fn revoke(
        &self,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
        token: &CompletionToken,
    ) -> Result<(), ServerError> {
        let mut state = self.state()?;
        let Entry::Occupied(mut slot) = state.entry((workflow_id.clone(), activity_id.clone()))
        else {
            return Ok(());
        };
        slot.get_mut().outstanding.retain(|issued| issued != token);
        if slot.get().outstanding.is_empty() {
            slot.remove();
        }
        Ok(())
    }

    /// Revoke whichever generation is current while parking for recovery.
    ///
    /// A park retires the execution site deliberately, so it takes every token
    /// of the current generation with it — including a redelivery's sibling.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] when fence state cannot be trusted.
    pub fn revoke_current(
        &self,
        workflow_id: &WorkflowId,
        activity_id: &ActivityId,
    ) -> Result<(), ServerError> {
        self.state()?
            .remove(&(workflow_id.clone(), activity_id.clone()));
        Ok(())
    }

    /// Mark `token`'s lease append as in flight: a completion for it waits at
    /// [`Self::lease_settled`] until [`Self::settle_lease`] runs.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the gate lock is poisoned.
    pub fn arm_lease(&self, token: &CompletionToken) -> Result<(), ServerError> {
        self.gate()?.insert(token.as_str().to_owned());
        Ok(())
    }

    /// The lease append for `token` has landed, failed, or been abandoned —
    /// release every completion waiting on it.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the gate lock is poisoned.
    pub fn settle_lease(&self, token: &CompletionToken) -> Result<(), ServerError> {
        self.gate()?.remove(token.as_str());
        self.lease_gate.settled.notify_waiters();
        Ok(())
    }

    /// Whether `token`'s lease append is still in flight.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the gate lock is poisoned.
    pub fn lease_pending(&self, token: &CompletionToken) -> Result<bool, ServerError> {
        Ok(self.gate()?.contains(token.as_str()))
    }

    /// Wait until `token`'s lease append is no longer in flight. Returns at
    /// once for a token that was never armed or is already settled.
    ///
    /// The completion entry points await this before handing a result to the
    /// fence; [`Self::accept`] also waits when it can host the wait itself.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if the gate lock is poisoned.
    pub async fn lease_settled(&self, token: &CompletionToken) -> Result<(), ServerError> {
        loop {
            // Register interest BEFORE the check, so a settle that lands
            // between the check and the wait is not missed.
            let notified = self.lease_gate.settled.notified();
            if !self.lease_pending(token)? {
                return Ok(());
            }
            notified.await;
        }
    }

    /// [`Self::lease_settled`] from the synchronous fence: hosted with
    /// `block_in_place` inside a multi-thread runtime; on a plain thread the
    /// wait runs on a private current-thread runtime. Inside a current-thread
    /// runtime nothing can host a blocking wait without starving the task that
    /// settles the gate, so the fence relies on the async entry point having
    /// awaited already and proceeds — and says so at debug level.
    fn wait_lease_settled_blocking(&self, token: &CompletionToken) -> Result<(), ServerError> {
        if !self.lease_pending(token)? {
            return Ok(());
        }
        if let Ok(handle) = tokio::runtime::Handle::try_current() {
            return match handle.runtime_flavor() {
                tokio::runtime::RuntimeFlavor::MultiThread => {
                    tokio::task::block_in_place(|| handle.block_on(self.lease_settled(token)))
                }
                flavor => {
                    // The one place the fence's promise thins. Said at WARN,
                    // not DEBUG: current-thread is what `#[tokio::test]`
                    // defaults to, so a test driving the sync entry could
                    // otherwise pass green-by-construction with no visible
                    // trace that the gate was never waited on. Not counted on
                    // the lease-loss ledger: no lease record is lost here —
                    // the ordering guarantee is what is being relied on.
                    tracing::warn!(
                        token = token.as_str(),
                        ?flavor,
                        "lease gate cannot be waited on synchronously inside this runtime; the \
                         async completion entry is relied on to have awaited it"
                    );
                    Ok(())
                }
            };
        }
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .map_err(|error| {
                ServerError::worker_dispatch(
                    "",
                    "",
                    format!("lease gate wait could not build a runtime: {error}"),
                )
            })?;
        runtime.block_on(self.lease_settled(token))
    }

    fn gate(&self) -> Result<MutexGuard<'_, HashSet<String>>, ServerError> {
        self.lease_gate
            .pending
            .lock()
            .map_err(|_| ServerError::lock_poisoned("activity lease gate"))
    }

    fn state(&self) -> Result<MutexGuard<'_, HashMap<ExecutionKey, SiteGeneration>>, ServerError> {
        self.current
            .lock()
            .map_err(|_| ServerError::lock_poisoned("activity completion fences"))
    }
}

/// Derive the stable external-effect key for one action site in one workflow run.
///
/// Attempts and execution generations are intentionally absent. The domain tag,
/// workflow id, run id, and activity ordinal are length-unambiguous fixed-width
/// inputs to SHA-256.
#[must_use]
pub fn idempotency_key(
    workflow_id: &WorkflowId,
    run_id: &RunId,
    activity_id: &ActivityId,
) -> String {
    let mut hasher = Sha256::new();
    hasher.update(b"aion.activity.idempotency.v1\0");
    hasher.update(workflow_id.as_uuid().as_bytes());
    hasher.update(run_id.as_uuid().as_bytes());
    hasher.update(activity_id.sequence_position().to_be_bytes());
    encode_hex(&hasher.finalize())
}

fn encode_hex(bytes: &[u8]) -> String {
    const DIGITS: &[u8; 16] = b"0123456789abcdef";
    let mut encoded = String::with_capacity(bytes.len() * 2);
    for byte in bytes {
        encoded.push(char::from(DIGITS[usize::from(byte >> 4)]));
        encoded.push(char::from(DIGITS[usize::from(byte & 0x0f)]));
    }
    encoded
}

fn rejection(
    workflow_id: &WorkflowId,
    activity_id: &ActivityId,
    reason: CompletionRejectionReason,
) -> ServerError {
    ServerError::ActivityCompletionRejected {
        workflow_id: workflow_id.clone(),
        activity_id: activity_id.clone(),
        reason,
    }
}

#[cfg(test)]
mod tests {

    /// WA-010 R3 ordering pin: a completion that arrives while its lease is
    /// still being appended is accepted only AFTER the lease settles — the
    /// fence hosts the wait itself on a multi-thread runtime.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_completion_racing_its_lease_is_accepted_after_the_lease_settles()
    -> Result<(), ServerError> {
        use std::sync::atomic::{AtomicU64, Ordering};
        let fences = CompletionFences::default();
        let workflow = WorkflowId::new(Uuid::new_v4());
        let run = RunId::new(Uuid::new_v4());
        let activity = ActivityId::from_sequence_position(1);
        let token = fences.issue(&workflow, &run, &activity, 1)?;
        fences.arm_lease(&token)?;
        let clock = Arc::new(AtomicU64::new(0));
        let lease_landed = Arc::new(AtomicU64::new(0));
        let accepted_at = Arc::new(AtomicU64::new(0));
        let racing = {
            let fences = fences.clone();
            let workflow = workflow.clone();
            let activity = activity.clone();
            let token = token.clone();
            let clock = Arc::clone(&clock);
            let accepted_at = Arc::clone(&accepted_at);
            tokio::spawn(async move {
                let accepted = fences.accept(&workflow, &activity, &token);
                accepted_at.store(clock.fetch_add(1, Ordering::SeqCst) + 1, Ordering::SeqCst);
                accepted.map(|_| ())
            })
        };
        // The completion is in flight; the lease "lands" now.
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        assert_eq!(
            accepted_at.load(Ordering::SeqCst),
            0,
            "the fence must not have accepted the completion while the lease was pending"
        );
        lease_landed.store(clock.fetch_add(1, Ordering::SeqCst) + 1, Ordering::SeqCst);
        fences.settle_lease(&token)?;
        racing
            .await
            .map_err(|error| ServerError::worker_dispatch("", "", error.to_string()))??;
        assert!(
            accepted_at.load(Ordering::SeqCst) > lease_landed.load(Ordering::SeqCst),
            "the completion was accepted before its lease landed: accepted at {} vs lease at {}",
            accepted_at.load(Ordering::SeqCst),
            lease_landed.load(Ordering::SeqCst)
        );
        Ok(())
    }

    #[tokio::test]
    async fn an_unarmed_token_never_waits_and_a_settled_one_is_released() -> Result<(), ServerError>
    {
        let fences = CompletionFences::default();
        let workflow = WorkflowId::new(Uuid::new_v4());
        let run = RunId::new(Uuid::new_v4());
        let activity = ActivityId::from_sequence_position(1);
        let token = fences.issue(&workflow, &run, &activity, 1)?;
        fences.lease_settled(&token).await?;
        fences.arm_lease(&token)?;
        assert!(fences.lease_pending(&token)?);
        let waiter = {
            let fences = fences.clone();
            let token = token.clone();
            tokio::spawn(async move { fences.lease_settled(&token).await })
        };
        fences.settle_lease(&token)?;
        waiter
            .await
            .map_err(|error| ServerError::worker_dispatch("", "", error.to_string()))??;
        assert!(!fences.lease_pending(&token)?);
        Ok(())
    }
    use super::{CompletionFences, CompletionToken, idempotency_key};
    use crate::error::{CompletionRejectionReason, ServerError};
    use aion_core::{ActivityId, RunId, WorkflowId};
    use std::sync::Arc;
    use uuid::Uuid;

    type TestResult = Result<(), Box<dyn std::error::Error>>;

    #[test]
    fn idempotency_key_is_attempt_independent_and_site_run_scoped() {
        let workflow = WorkflowId::new_v4();
        let run_a = RunId::new_v4();
        let run_b = RunId::new_v4();
        let site_a = ActivityId::from_sequence_position(7);
        let site_b = ActivityId::from_sequence_position(8);

        let first_attempt = idempotency_key(&workflow, &run_a, &site_a);
        let fifth_attempt = idempotency_key(&workflow, &run_a, &site_a);
        assert_eq!(first_attempt, fifth_attempt);
        assert_ne!(first_attempt, idempotency_key(&workflow, &run_a, &site_b));
        assert_ne!(first_attempt, idempotency_key(&workflow, &run_b, &site_a));
    }

    /// The FIRST attempt's token is refused once a genuine RETRY attempt has
    /// been issued: c4e1412d7's invariant, by the attempt discriminator it put
    /// on the envelope.
    #[test]
    fn issuing_a_retry_rejects_the_stale_generation() -> TestResult {
        let fences = CompletionFences::default();
        let workflow = WorkflowId::new_v4();
        let run = RunId::new_v4();
        let activity = ActivityId::from_sequence_position(3);
        let stale = fences.issue(&workflow, &run, &activity, 1)?;
        let current = fences.issue(&workflow, &run, &activity, 2)?;

        let rejected = fences.accept(&workflow, &activity, &stale);
        assert!(matches!(
            rejected,
            Err(ServerError::ActivityCompletionRejected {
                reason: CompletionRejectionReason::StaleGeneration,
                ..
            })
        ));
        fences.accept(&workflow, &activity, &current)?;
        Ok(())
    }

    #[test]
    fn accepted_generation_is_consumed_exactly_once() -> TestResult {
        let fences = CompletionFences::default();
        let workflow = WorkflowId::new_v4();
        let run = RunId::new_v4();
        let activity = ActivityId::from_sequence_position(4);
        let token = fences.issue(&workflow, &run, &activity, 1)?;

        fences.accept(&workflow, &activity, &token)?;
        let duplicate = fences.accept(&workflow, &activity, &token);
        assert!(matches!(
            duplicate,
            Err(ServerError::ActivityCompletionRejected {
                reason: CompletionRejectionReason::NoCurrentGeneration,
                ..
            })
        ));
        Ok(())
    }

    /// Revoking a superseded ATTEMPT's token never disturbs the retry that
    /// replaced it. The same-attempt sibling case this name never claimed is
    /// owned by
    /// [`the_all_streams_closed_revoke_cannot_orphan_a_redelivered_sibling`].
    #[test]
    fn revoking_an_old_generation_does_not_remove_its_replacement() -> TestResult {
        let fences = CompletionFences::default();
        let workflow = WorkflowId::new_v4();
        let run = RunId::new_v4();
        let activity = ActivityId::from_sequence_position(6);
        let old = fences.issue(&workflow, &run, &activity, 1)?;
        let replacement = fences.issue(&workflow, &run, &activity, 2)?;

        fences.revoke(&workflow, &activity, &old)?;
        fences.accept(&workflow, &activity, &replacement)?;
        Ok(())
    }

    #[test]
    fn an_empty_wire_token_is_a_typed_compatibility_refusal() {
        let workflow = WorkflowId::new_v4();
        let activity = ActivityId::from_sequence_position(9);
        let rejected = CompletionToken::from_wire(&workflow, &activity, String::new());
        assert!(matches!(
            rejected,
            Err(ServerError::ActivityCompletionRejected {
                reason: CompletionRejectionReason::MissingCompletionToken,
                ..
            })
        ));
    }

    #[test]
    fn a_pre_recovery_generation_is_rejected_after_recovery() -> TestResult {
        let before_recovery = CompletionFences::default();
        let workflow = WorkflowId::new_v4();
        let run = RunId::new_v4();
        let activity = ActivityId::from_sequence_position(5);
        let stale = before_recovery.issue(&workflow, &run, &activity, 1)?;

        let after_recovery = CompletionFences::default();
        let current = after_recovery.issue(&workflow, &run, &activity, 1)?;
        let rejected = after_recovery.accept(&workflow, &activity, &stale);
        assert!(matches!(
            rejected,
            Err(ServerError::ActivityCompletionRejected {
                reason: CompletionRejectionReason::StaleGeneration,
                ..
            })
        ));
        after_recovery.accept(&workflow, &activity, &current)?;
        Ok(())
    }

    /// FENCE-1 T1: a redelivery between delivery and completion does not orphan
    /// the finished result.
    ///
    /// Delivery is at-least-once (`transport_loss.rs`: worker loss is
    /// attempt-neutral and re-dispatches the SAME attempt), so a worker that is
    /// alive and finishing can have its work delivered a second time. The first
    /// worker still holds the first token and its result is real: it is
    /// accepted. The redelivery's token is then the duplicate, and it is refused
    /// because the first acceptance consumed the whole generation.
    ///
    /// On the base this reads `StaleGeneration` for the FIRST worker — the
    /// finished result is thrown away, and nothing retries because the outbox
    /// row settled `Done` at dispatch.
    #[test]
    fn a_redelivery_of_the_same_attempt_does_not_orphan_the_first_worker() -> TestResult {
        let fences = CompletionFences::default();
        let workflow = WorkflowId::new_v4();
        let run = RunId::new_v4();
        let activity = ActivityId::from_sequence_position(11);

        let first = fences.issue(&workflow, &run, &activity, 1)?;
        // The redelivery: the SAME attempt of the SAME run, dispatched again.
        let second = fences.issue(&workflow, &run, &activity, 1)?;

        let accepted = fences.accept(&workflow, &activity, &first);
        assert!(
            accepted.is_ok(),
            "the worker that genuinely held the FIRST delivery finished the work; its result \
             must be accepted, not thrown away: {accepted:?}"
        );

        let duplicate = fences.accept(&workflow, &activity, &second);
        assert!(
            matches!(
                duplicate,
                Err(ServerError::ActivityCompletionRejected {
                    reason: CompletionRejectionReason::NoCurrentGeneration,
                    ..
                })
            ),
            "the first accepted completion must consume EVERY outstanding token for the site, \
             so the redelivered worker's completion is the duplicate: {duplicate:?}"
        );
        Ok(())
    }

    /// FENCE-1 T1b: the all-streams-closed revoke
    /// (`dispatch.rs` `send_to_candidates`, immediately before `Ok(None)`)
    /// cannot orphan a sibling still held by a live worker.
    ///
    /// This is the exact rocketfish 09:57:54Z interleaving: attempt A is
    /// delivered and held, the redelivery B is issued for the same attempt,
    /// every candidate stream is then found closed so B revokes its OWN token —
    /// and on the base that removed the site's only key, so A's completion read
    /// `NoCurrentGeneration` rather than `StaleGeneration`.
    #[test]
    fn the_all_streams_closed_revoke_cannot_orphan_a_redelivered_sibling() -> TestResult {
        let fences = CompletionFences::default();
        let workflow = WorkflowId::new_v4();
        let run = RunId::new_v4();
        let activity = ActivityId::from_sequence_position(12);

        let delivered = fences.issue(&workflow, &run, &activity, 1)?;
        let redelivery = fences.issue(&workflow, &run, &activity, 1)?;

        // Every candidate stream was closed, so the redelivery withdraws the
        // authorization IT minted — and only that one.
        fences.revoke(&workflow, &activity, &redelivery)?;

        let accepted = fences.accept(&workflow, &activity, &delivered);
        assert!(
            accepted.is_ok(),
            "the sibling token the first worker still holds must survive the redelivery's own \
             revoke: {accepted:?}"
        );

        let withdrawn = fences.accept(&workflow, &activity, &redelivery);
        assert!(
            matches!(
                withdrawn,
                Err(ServerError::ActivityCompletionRejected {
                    reason: CompletionRejectionReason::NoCurrentGeneration,
                    ..
                })
            ),
            "a revoked token must never become truth: {withdrawn:?}"
        );
        Ok(())
    }

    /// FENCE-1 T2, half one: a superseded ATTEMPT is still refused.
    ///
    /// Regression pin, not a red-first test: it is green on the base too, for
    /// the wrong reason (the base replaced the single slot on every issue). Its
    /// value is that it stays green through the change, which is what says
    /// c4e1412d7's invariant kept its name.
    #[test]
    fn a_superseded_attempt_is_still_refused_after_a_retry_is_issued() -> TestResult {
        let fences = CompletionFences::default();
        let workflow = WorkflowId::new_v4();
        let run = RunId::new_v4();
        let activity = ActivityId::from_sequence_position(13);

        let earlier = fences.issue(&workflow, &run, &activity, 1)?;
        let current = fences.issue(&workflow, &run, &activity, 2)?;

        let refused = fences.accept(&workflow, &activity, &earlier);
        assert!(
            matches!(
                refused,
                Err(ServerError::ActivityCompletionRejected {
                    reason: CompletionRejectionReason::StaleGeneration,
                    ..
                })
            ),
            "a stale worker can never be recorded as truth: attempt 1's worker was superseded by \
             attempt 2 and its completion must stay refused: {refused:?}"
        );
        fences.accept(&workflow, &activity, &current)?;
        Ok(())
    }

    /// FENCE-1 T2, half two: the EXECUTION-GENERATION boundary is still
    /// refused across, with identical attempt numbers on both sides.
    ///
    /// A reset or continue-as-new mints a new [`RunId`], so the run-scoped
    /// idempotency key c4e1412d7 already derives is the only discriminator that
    /// separates these two attempt-1 tokens. This is the case
    /// acceptance-by-held-generation would be blind to if the key were not used.
    #[test]
    fn a_superseded_execution_generation_is_still_refused_at_the_same_attempt() -> TestResult {
        let fences = CompletionFences::default();
        let workflow = WorkflowId::new_v4();
        let run_a = RunId::new_v4();
        let run_b = RunId::new_v4();
        let activity = ActivityId::from_sequence_position(14);

        let old_run = fences.issue(&workflow, &run_a, &activity, 1)?;
        let new_run = fences.issue(&workflow, &run_b, &activity, 1)?;

        let refused = fences.accept(&workflow, &activity, &old_run);
        assert!(
            matches!(
                refused,
                Err(ServerError::ActivityCompletionRejected {
                    reason: CompletionRejectionReason::StaleGeneration,
                    ..
                })
            ),
            "a stale worker can never be recorded as truth: the superseded RUN's worker holds \
             attempt 1 of a generation that no longer exists: {refused:?}"
        );
        fences.accept(&workflow, &activity, &new_run)?;
        Ok(())
    }

    /// A settlement that fails after acceptance restores the WHOLE generation,
    /// siblings included, so the true resolver — whichever token it holds — is
    /// not refused with `NoCurrentGeneration`.
    #[test]
    fn a_restored_generation_carries_its_redelivered_sibling_back() -> TestResult {
        let fences = CompletionFences::default();
        let workflow = WorkflowId::new_v4();
        let run = RunId::new_v4();
        let activity = ActivityId::from_sequence_position(15);

        let delivered = fences.issue(&workflow, &run, &activity, 1)?;
        let redelivery = fences.issue(&workflow, &run, &activity, 1)?;

        let accepted = fences.accept(&workflow, &activity, &delivered)?;
        assert_eq!(accepted.attempt(), 1);
        fences.restore_if_absent(&workflow, &activity, &accepted)?;

        // The true resolver presents the sibling token and is accepted.
        fences.accept(&workflow, &activity, &redelivery)?;
        Ok(())
    }

    /// A restore never overwrites a generation issued while the settlement was
    /// in flight: a concurrently issued newer generation always wins.
    #[test]
    fn a_restore_never_displaces_a_newer_generation() -> TestResult {
        let fences = CompletionFences::default();
        let workflow = WorkflowId::new_v4();
        let run = RunId::new_v4();
        let activity = ActivityId::from_sequence_position(16);

        let first = fences.issue(&workflow, &run, &activity, 1)?;
        let accepted = fences.accept(&workflow, &activity, &first)?;
        let retry = fences.issue(&workflow, &run, &activity, 2)?;

        fences.restore_if_absent(&workflow, &activity, &accepted)?;

        let refused = fences.accept(&workflow, &activity, &first);
        assert!(
            matches!(
                refused,
                Err(ServerError::ActivityCompletionRejected {
                    reason: CompletionRejectionReason::StaleGeneration,
                    ..
                })
            ),
            "the restore must not displace the retry that was issued meanwhile: {refused:?}"
        );
        fences.accept(&workflow, &activity, &retry)?;
        Ok(())
    }

    /// A re-dispatch of an ALREADY superseded attempt registers nothing: it
    /// cannot resurrect the attempt the engine has moved past, and the token it
    /// hands back is refused when presented.
    #[test]
    fn re_issuing_a_superseded_attempt_registers_no_generation() -> TestResult {
        let fences = CompletionFences::default();
        let workflow = WorkflowId::new_v4();
        let run = RunId::new_v4();
        let activity = ActivityId::from_sequence_position(17);

        let current = fences.issue(&workflow, &run, &activity, 2)?;
        let out_of_order = fences.issue(&workflow, &run, &activity, 1)?;

        let refused = fences.accept(&workflow, &activity, &out_of_order);
        assert!(
            matches!(
                refused,
                Err(ServerError::ActivityCompletionRejected {
                    reason: CompletionRejectionReason::StaleGeneration,
                    ..
                })
            ),
            "an out-of-order re-dispatch of a superseded attempt must not become acceptable: \
             {refused:?}"
        );
        fences.accept(&workflow, &activity, &current)?;
        Ok(())
    }

    /// Two workers holding sibling tokens for one redelivered attempt: exactly
    /// one becomes truth. The set of tokens that MAY be first is wider; how
    /// many may be first is not.
    #[test]
    fn only_one_of_two_sibling_tokens_can_ever_become_truth() -> TestResult {
        let fences = CompletionFences::default();
        let workflow = WorkflowId::new_v4();
        let run = RunId::new_v4();
        let activity = ActivityId::from_sequence_position(18);

        let first = fences.issue(&workflow, &run, &activity, 1)?;
        let second = fences.issue(&workflow, &run, &activity, 1)?;

        let accepted = [
            fences.accept(&workflow, &activity, &second).is_ok(),
            fences.accept(&workflow, &activity, &first).is_ok(),
        ];
        assert_eq!(
            accepted.iter().filter(|ok| **ok).count(),
            1,
            "exactly one completion for a redelivered attempt may become truth"
        );
        Ok(())
    }
}