fix(engine): release trigger provider guards before await - #2198

// Real teardown order (engine/src/engine/mod.rs, handle_worker):
//     writer.abort();                 // drops the channel Receiver
//     self.cleanup_worker(&worker)    // -> trigger_registry.unregister_worker
// so every send to a departed provider fails with "channel closed".
#[cfg(test)]
mod review_2198_portable {
    use super::*;
    use crate::engine::Outbound;
    use crate::protocol::{DEFAULT_NAMESPACE, Message};
    use crate::worker_connections::WorkerConnection;
    use std::collections::HashSet;
    use std::sync::Arc;

    #[derive(Default)]
    pub(super) struct Probe {
        pub live: std::sync::Mutex<HashSet<String>>,
        pub events: std::sync::Mutex<Vec<String>>,
    }

    impl TriggerRegistrator for Arc<Probe> {
        fn register_trigger(
            &self,
            trigger: Trigger,
        ) -> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + '_>> {
            Box::pin(async move {
                self.live.lock().unwrap().insert(trigger.id.clone());
                self.events.lock().unwrap().push(format!("+{}", trigger.id));
                Ok(())
            })
        }

        fn unregister_trigger(
            &self,
            trigger: Trigger,
        ) -> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + '_>> {
            Box::pin(async move {
                self.live.lock().unwrap().remove(&trigger.id);
                self.events.lock().unwrap().push(format!("-{}", trigger.id));
                Ok(())
            })
        }
    }

    pub(super) fn base_trigger(id: &str) -> Trigger {
        Trigger {
            id: id.to_string(),
            trigger_type: "evt".to_string(),
            function_id: format!("fn_{id}"),
            config: serde_json::json!({}),
            worker_id: None,
            metadata: None,
            namespace: DEFAULT_NAMESPACE.to_string(),
            trigger_namespace: None,
            home_namespace: DEFAULT_NAMESPACE.to_string(),
            provider_namespace: DEFAULT_NAMESPACE.to_string(),
        }
    }

    /// A binding sent by a consumer worker (connection-owned, fire-and-forget).
    pub(super) fn consumer_trigger(id: &str, owner: Uuid) -> Trigger {
        Trigger {
            worker_id: Some(owner),
            ..base_trigger(id)
        }
    }

    pub(super) fn worker_provider(
        capacity: usize,
    ) -> (WorkerConnection, tokio::sync::mpsc::Receiver<Outbound>) {
        let (tx, rx) = tokio::sync::mpsc::channel(capacity);
        (WorkerConnection::new(tx), rx)
    }

    /// Mirrors router_msg's RegisterTriggerType: a fresh Box per message.
    pub(super) async fn install_worker_provider(
        registry: &TriggerRegistry,
        namespace: &str,
        connection: &WorkerConnection,
    ) {
        registry
            .register_trigger_type(TriggerType::new_ns(
                namespace,
                "evt",
                "worker provider",
                Box::new(connection.clone()),
                Some(connection.id),
            ))
            .await
            .unwrap();
    }

    pub(super) fn drain(rx: &mut tokio::sync::mpsc::Receiver<Outbound>) -> Vec<String> {
        let mut out = Vec::new();
        while let Ok(msg) = rx.try_recv() {
            out.push(match msg {
                Outbound::Protocol(Message::RegisterTrigger { id, .. }) => format!("+{id}"),
                Outbound::Protocol(Message::UnregisterTrigger { id, .. }) => format!("-{id}"),
                _ => "?".to_string(),
            });
        }
        out
    }

    /// Provider P1 dies (writer aborted, then cleanup) and reconnects as P2.
    /// Returns P2 and its receiver with the replay already drained.
    pub(super) async fn provider_restart_with_binding(
        registry: &TriggerRegistry,
        consumer: Uuid,
    ) -> (WorkerConnection, tokio::sync::mpsc::Receiver<Outbound>) {
        let (p1, rx1) = worker_provider(8);
        install_worker_provider(registry, DEFAULT_NAMESPACE, &p1).await;
        registry
            .register_trigger(consumer_trigger("t", consumer))
            .await
            .unwrap();
        drop(rx1);
        registry.unregister_worker(&p1.id).await;
        assert!(registry.pending_triggers.contains_key("t"));

        let (p2, mut rx2) = worker_provider(8);
        install_worker_provider(registry, DEFAULT_NAMESPACE, &p2).await;
        assert!(registry.triggers.contains_key("t"));
        assert_eq!(drain(&mut rx2), vec!["+t"]);
        (p2, rx2)
    }

    /// R1 — provider restart (the exact #2178 flow), then the consumer
    /// unregisters. Expected: Ok(true), binding gone, P2 told "-t" only.
    #[tokio::test]
    async fn r1_unregister_after_provider_restart_really_removes_binding() {
        let registry = TriggerRegistry::new();
        let (_p2, mut rx2) = provider_restart_with_binding(&registry, Uuid::new_v4()).await;

        let result = registry.unregister_trigger("t".into(), None).await;
        let sent = drain(&mut rx2);
        assert!(
            result.is_ok() && !registry.triggers.contains_key("t") && sent == vec!["-t"],
            "unregister={result:?} still_live={} provider_saw={sent:?}",
            registry.triggers.contains_key("t")
        );
    }

    /// R2 — the provider is down, the binding is parked; the consumer
    /// unregisters it. Expected: Ok(true) and NOTHING replayed when the
    /// provider comes back.
    #[tokio::test]
    async fn r2_unregister_parked_intent_while_provider_is_down() {
        let registry = TriggerRegistry::new();
        let (p1, rx1) = worker_provider(8);
        install_worker_provider(&registry, DEFAULT_NAMESPACE, &p1).await;
        registry
            .register_trigger(consumer_trigger("t", Uuid::new_v4()))
            .await
            .unwrap();
        drop(rx1);
        registry.unregister_worker(&p1.id).await;
        assert!(registry.pending_triggers.contains_key("t"));

        let result = registry.unregister_trigger("t".into(), None).await;
        let still_parked = registry.pending_triggers.contains_key("t");

        let (p2, mut rx2) = worker_provider(8);
        install_worker_provider(&registry, DEFAULT_NAMESPACE, &p2).await;
        let replayed = drain(&mut rx2);
        assert!(
            result.is_ok()
                && !still_parked
                && replayed.is_empty()
                && !registry.triggers.contains_key("t"),
            "unregister={result:?} still_parked={still_parked} replayed_on_return={replayed:?} live_again={}",
            registry.triggers.contains_key("t")
        );
    }

    /// R3 — no restart at all: a home provider leaves, the binding falls back
    /// to the DEFAULT provider, then the consumer unregisters it.
    #[tokio::test]
    async fn r3_unregister_after_home_provider_left_for_fallback() {
        let registry = TriggerRegistry::new();
        let fallback = Arc::new(Probe::default());
        registry
            .register_trigger_type(TriggerType::new_ns(
                DEFAULT_NAMESPACE,
                "evt",
                "fallback",
                Box::new(Arc::clone(&fallback)),
                None,
            ))
            .await
            .unwrap();
        let (home, rx_home) = worker_provider(8);
        install_worker_provider(&registry, "shop", &home).await;
        let binding = Trigger {
            home_namespace: "shop".into(),
            ..consumer_trigger("t", Uuid::new_v4())
        };
        registry.register_trigger(binding).await.unwrap();
        assert_eq!(registry.triggers.get("t").unwrap().provider_namespace, "shop");

        drop(rx_home);
        registry.unregister_worker(&home.id).await;
        assert!(fallback.live.lock().unwrap().contains("t"));

        let result = registry.unregister_trigger("t".into(), None).await;
        let events = fallback.events.lock().unwrap().clone();
        assert!(
            result.is_ok() && fallback.live.lock().unwrap().is_empty(),
            "unregister={result:?} fallback_still_runs={} fallback_events={events:?}",
            fallback.live.lock().unwrap().contains("t")
        );
    }

    /// R4 — the SAME connection re-announcing its trigger type (router builds
    /// a fresh Box each time) must not tear bindings down. main replays "+t";
    /// a detach first opens a window where the provider has no job.
    #[tokio::test]
    async fn r4_same_connection_reregistration_does_not_detach() {
        let registry = TriggerRegistry::new();
        let (p, mut rx) = worker_provider(8);
        install_worker_provider(&registry, DEFAULT_NAMESPACE, &p).await;
        registry
            .register_trigger(consumer_trigger("t", Uuid::new_v4()))
            .await
            .unwrap();
        drain(&mut rx);
        install_worker_provider(&registry, DEFAULT_NAMESPACE, &p).await;
        let sent = drain(&mut rx);
        assert!(!sent.contains(&"-t".to_string()), "same connection saw {sent:?}");
    }

    /// R5 — re-registering an id onto another provider (route change) must
    /// retire the job on the previous provider; otherwise both fire.
    /// (Pre-existing on main, where the old job is even untracked.)
    #[tokio::test]
    async fn r5_route_change_detaches_previous_provider() {
        let registry = TriggerRegistry::new();
        let old = Arc::new(Probe::default());
        let other = Arc::new(Probe::default());
        for (ns, probe) in [(DEFAULT_NAMESPACE, &old), ("other", &other)] {
            registry
                .register_trigger_type(TriggerType::new_ns(
                    ns,
                    "evt",
                    "probe",
                    Box::new(Arc::clone(probe)),
                    None,
                ))
                .await
                .unwrap();
        }
        registry.register_trigger(base_trigger("t")).await.unwrap();
        let moved = Trigger {
            trigger_namespace: Some("other".into()),
            ..base_trigger("t")
        };
        registry.register_trigger(moved).await.unwrap();
        assert!(other.live.lock().unwrap().contains("t"));
        assert!(
            old.live.lock().unwrap().is_empty(),
            "old provider still runs superseded binding: {:?}",
            old.events.lock().unwrap()
        );
    }
}

// Needs PR-only internals (lifecycles, TRIGGER_OPERATION_TIMEOUT).
#[cfg(test)]
mod review_2198_pr_only {
    use super::review_2198_portable::*;
    use super::*;

    fn retained_deliveries(registry: &TriggerRegistry, id: &str) -> Option<usize> {
        registry.lifecycles.get(id).map(|gate| {
            gate.try_lock()
                .map(|s| s.deliveries.len() + s.uncertain.len())
                .unwrap_or(usize::MAX)
        })
    }

    /// R6 — the consumer disconnects after a provider restart. Both maps are
    /// cleared, but the gate (holding the dead WorkerConnection) must go too.
    #[tokio::test]
    async fn r6_owner_disconnect_after_provider_restart_reclaims_gate() {
        let registry = TriggerRegistry::new();
        let consumer = Uuid::new_v4();
        let (_p2, _rx2) = provider_restart_with_binding(&registry, consumer).await;
        registry.unregister_worker(&consumer).await;
        assert!(registry.triggers.is_empty());
        assert!(registry.pending_triggers.is_empty());
        assert!(
            registry.lifecycles.is_empty(),
            "gate leaked with {:?} retained deliveries",
            retained_deliveries(&registry, "t")
        );
    }

    /// R7 — every provider restart appends one more dead generation to the
    /// binding's delivery list (each one retains a WorkerConnection clone).
    #[tokio::test]
    async fn r7_retained_generations_do_not_grow_with_restarts() {
        let registry = TriggerRegistry::new();
        let (mut provider, mut rx) = worker_provider(8);
        install_worker_provider(&registry, crate::protocol::DEFAULT_NAMESPACE, &provider).await;
        registry
            .register_trigger(consumer_trigger("t", Uuid::new_v4()))
            .await
            .unwrap();
        for _ in 0..5 {
            drop(rx);
            registry.unregister_worker(&provider.id).await;
            let (next, next_rx) = worker_provider(8);
            install_worker_provider(&registry, crate::protocol::DEFAULT_NAMESPACE, &next).await;
            provider = next;
            rx = next_rx;
        }
        let retained = retained_deliveries(&registry, "t");
        assert!(
            retained <= Some(1),
            "after 5 restarts the binding retains {retained:?} provider generations"
        );
    }

    /// Registration succeeds; detach never completes (e.g. provider stopped
    /// draining its socket).
    struct StuckDetach;

    impl TriggerRegistrator for StuckDetach {
        fn register_trigger(
            &self,
            _trigger: Trigger,
        ) -> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + '_>> {
            Box::pin(async { Ok(()) })
        }

        fn unregister_trigger(
            &self,
            _trigger: Trigger,
        ) -> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + '_>> {
            Box::pin(std::future::pending())
        }
    }

    /// R8 — the 10 s budget is per binding, not per operation: a consumer with
    /// N bindings on a stalled provider blocks cleanup_worker for N x 10 s.
    #[tokio::test(start_paused = true)]
    async fn r8_owner_cleanup_is_not_linear_in_bindings() {
        const N: u32 = 30;
        let registry = TriggerRegistry::new();
        registry
            .register_trigger_type(TriggerType::new(
                "evt",
                "stuck",
                Box::new(StuckDetach),
                Some(Uuid::new_v4()),
            ))
            .await
            .unwrap();
        let owner = Uuid::new_v4();
        for i in 0..N {
            registry
                .register_trigger(consumer_trigger(&format!("t{i}"), owner))
                .await
                .unwrap();
        }
        let start = tokio::time::Instant::now();
        registry.unregister_worker(&owner).await;
        let elapsed = tokio::time::Instant::now() - start;
        assert!(
            elapsed <= TRIGGER_OPERATION_TIMEOUT,
            "owner cleanup took {elapsed:?} (virtual) for {N} bindings"
        );
    }

    #[derive(Default)]
    struct PausingProbe {
        pause: std::sync::atomic::AtomicBool,
        resume: tokio::sync::Notify,
    }

    impl TriggerRegistrator for std::sync::Arc<PausingProbe> {
        fn register_trigger(
            &self,
            _trigger: Trigger,
        ) -> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + '_>> {
            Box::pin(async move {
                if self.pause.swap(false, std::sync::atomic::Ordering::SeqCst) {
                    self.resume.notified().await;
                }
                Ok(())
            })
        }

        fn unregister_trigger(
            &self,
            _trigger: Trigger,
        ) -> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + '_>> {
            Box::pin(async { Ok(()) })
        }
    }

    /// R9 — variant of `disconnect_does_not_wait_for_unrelated_busy_binding`
    /// using RESOLVED bindings (what workers send by default) instead of
    /// explicit namespaces.
    #[tokio::test]
    async fn r9_default_provider_disconnect_waits_for_unrelated_home_binding() {
        let registry = TriggerRegistry::new();
        let default_owner = Uuid::new_v4();
        let shop = std::sync::Arc::new(PausingProbe::default());
        registry
            .register_trigger_type(TriggerType::new_ns(
                crate::protocol::DEFAULT_NAMESPACE,
                "evt",
                "default",
                Box::new(std::sync::Arc::new(PausingProbe::default())),
                Some(default_owner),
            ))
            .await
            .unwrap();
        registry
            .register_trigger_type(TriggerType::new_ns(
                "shop",
                "evt",
                "shop",
                Box::new(std::sync::Arc::clone(&shop)),
                Some(Uuid::new_v4()),
            ))
            .await
            .unwrap();
        shop.pause.store(true, std::sync::atomic::Ordering::SeqCst);
        let binding = Trigger {
            home_namespace: "shop".into(),
            ..consumer_trigger("shop-binding", Uuid::new_v4())
        };
        let registration = registry.register_trigger(binding);
        tokio::pin!(registration);
        assert!(futures::poll!(&mut registration).is_pending());
        let cleanup = registry.unregister_worker(&default_owner);
        tokio::pin!(cleanup);
        let ready = futures::poll!(&mut cleanup).is_ready();
        shop.resume.notify_one();
        registration.await.unwrap();
        cleanup.await;
        assert!(ready, "DEFAULT provider disconnect blocked on an unrelated shop binding");
    }

    /// R10 — head-of-line blocking: an ownerless registration holds the gate
    /// while waiting for P's ack; P's read loop then handles UnregisterTrigger
    /// for the same id and blocks on that gate until the 10 s fail-open.
    #[tokio::test(start_paused = true)]
    async fn r10_read_loop_blocks_behind_ack_only_it_can_deliver() {
        let registry = std::sync::Arc::new(TriggerRegistry::new());
        let (p, _rx) = worker_provider(8);
        install_worker_provider(&registry, crate::protocol::DEFAULT_NAMESPACE, &p).await;
        let holder = {
            let registry = std::sync::Arc::clone(&registry);
            tokio::spawn(async move { registry.register_trigger(base_trigger("t")).await })
        };
        for _ in 0..4 {
            tokio::task::yield_now().await;
        }
        assert_eq!(p.pending_trigger_acks.len(), 1, "holder should be awaiting the ack");
        let start = tokio::time::Instant::now();
        let _ = registry.unregister_trigger("t".into(), None).await;
        let waited = tokio::time::Instant::now() - start;
        holder.await.unwrap().unwrap();
        assert!(
            waited < std::time::Duration::from_secs(1),
            "provider read loop stalled {waited:?} behind an ack only it can read"
        );
    }
}