alpha2-native-gpu 0.5.3

//! Vulkan interop: the device the renderer shares with `FFmpeg`'s decoder and encoder, import
//! of decoded frames as wgpu plane views with layout transitions and timeline synchronization,
//! copies of rendered planes into encoder frames, and device capability probes.

use std::{
    collections::VecDeque,
    error::Error,
    ffi::{CStr, c_void},
    fmt,
    ptr::NonNull,
    sync::{
        Arc, Mutex, Weak,
        atomic::{AtomicBool, Ordering},
    },
};

use alpha2_ffmpeg_bridge::{
    BridgeError, VulkanDeviceInfo, VulkanFrameAccess, VulkanFrameAccessGuard, VulkanFrameLease,
    VulkanQueueLock, create_vulkan_device_context,
};
use alpha2_gpu::{ExternalQueueLock, GpuContext};
use ash::vk::{self, Handle};
use rsmpeg::avutil::AVHWDeviceContext;
use wgpu_hal::Instance as _;

use crate::{NativeGpuIdentity, NativePlaneLayout};

// Each fallible helper returns 0 or the failing step and writes that step's platform status to
// `status` (`include/alpha2_native_gpu_status.h`).
unsafe extern "C" {
    fn alpha2_native_gpu_vulkan_probe(access: *mut VulkanFrameAccess, status: *mut i32) -> i32;
    fn alpha2_native_gpu_vulkan_acquire_check(
        access: *const VulkanFrameAccess,
        status: *mut i32,
    ) -> i32;
    fn alpha2_native_gpu_vulkan_ticket_create(
        device: u64,
        allocator: u64,
        queue_family: u32,
        output: *mut *mut c_void,
        status: *mut i32,
    ) -> i32;
    fn alpha2_native_gpu_vulkan_ticket_poll(ticket: *const c_void, status: *mut i32) -> i32;
    fn alpha2_native_gpu_vulkan_ticket_reset(ticket: *mut c_void, status: *mut i32) -> i32;
    fn alpha2_native_gpu_vulkan_acquire_submit(
        access: *mut VulkanFrameAccess,
        ticket: *mut c_void,
        status: *mut i32,
    ) -> i32;
    fn alpha2_native_gpu_vulkan_release(
        access: *mut VulkanFrameAccess,
        ticket: *mut c_void,
        status: *mut i32,
    ) -> i32;
    fn alpha2_native_gpu_vulkan_signal(
        access: *mut VulkanFrameAccess,
        semaphore: u64,
        wait_value: u64,
        signal_value: u64,
        status: *mut i32,
    ) -> i32;
    fn alpha2_native_gpu_vulkan_import_destroy(ticket: *mut c_void);
    fn alpha2_native_gpu_vulkan_import_retire(ticket: *mut c_void, timeout_ns: u64) -> i32;
    #[cfg(test)]
    fn alpha2_native_gpu_vulkan_retained_helpers() -> u64;
    fn alpha2_native_gpu_vulkan_encode_copier_create(
        device: u64,
        queue_family: u32,
        allocator: u64,
        output: *mut *mut c_void,
        status: *mut i32,
    ) -> i32;
    fn alpha2_native_gpu_vulkan_encode_copier_destroy(copier: *mut c_void, status: *mut i32)
    -> i32;
    fn alpha2_native_gpu_vulkan_encode_copier_slots() -> u32;
    fn alpha2_native_gpu_vulkan_encode_copier_await_slot(
        copier: *mut c_void,
        slot: *mut u32,
        status: *mut i32,
    ) -> i32;
    fn alpha2_native_gpu_vulkan_encode_copier_submit(
        copier: *mut c_void,
        access: *mut VulkanFrameAccess,
        copy: *const VulkanEncodeCopy,
        status: *mut i32,
    ) -> i32;
    fn alpha2_native_gpu_vulkan_set_wait_timeout(nanoseconds: u64);
    fn alpha2_native_gpu_vulkan_wait_timeout() -> u64;
}

pub(crate) fn set_wait_timeout(timeout: std::time::Duration) {
    let nanoseconds = u64::try_from(timeout.as_nanos()).unwrap_or(u64::MAX);
    // SAFETY: stores one integer the helpers read on each wait.
    unsafe { alpha2_native_gpu_vulkan_set_wait_timeout(nanoseconds) };
}

fn wait_timeout() -> std::time::Duration {
    // SAFETY: reads one integer.
    std::time::Duration::from_nanos(unsafe { alpha2_native_gpu_vulkan_wait_timeout() })
}

fn helper_error(operation: &'static str, step: i32, status: i32) -> BridgeError {
    crate::status::helper_failure(
        alpha2_ffmpeg_bridge::NativeGpuBackend::Vulkan,
        operation,
        step,
        status,
    )
}

/// Fails the Nth call of an import helper on this thread, as a device fault would, so tests
/// reach the failure branches.
#[cfg(test)]
pub(crate) mod inject {
    use std::cell::Cell;
    use std::sync::atomic::{AtomicU64, Ordering};

    /// How many waits for given-back tickets have begun in this process.
    static RETIRES_BEGUN: AtomicU64 = AtomicU64::new(0);

    pub(super) fn retire_begins() {
        RETIRES_BEGUN.fetch_add(1, Ordering::Release);
    }

    /// How many waits for given-back tickets have begun in this process, on any thread.
    pub(crate) fn retires_begun() -> u64 {
        RETIRES_BEGUN.load(Ordering::Acquire)
    }

    #[derive(Clone, Copy, Debug)]
    pub(crate) enum Hook {
        AcquireSubmit,
        TicketPoll,
        TicketReset,
        Release,
    }

    thread_local! {
        static COUNTDOWNS: [Cell<u32>; 4] =
            const { [Cell::new(0), Cell::new(0), Cell::new(0), Cell::new(0)] };
        static STATUSES: [Cell<i32>; 4] =
            const { [Cell::new(0), Cell::new(0), Cell::new(0), Cell::new(0)] };
    }

    /// Fails the `nth` (1-based) next call of `hook` on this thread with a lost device.
    pub(crate) fn fail_nth(hook: Hook, nth: u32) {
        fail_nth_with(hook, nth, super::vk::Result::ERROR_DEVICE_LOST.as_raw());
    }

    /// Fails the `nth` (1-based) next call of `hook` on this thread with `VkResult` `status`.
    pub(crate) fn fail_nth_with(hook: Hook, nth: u32, status: i32) {
        COUNTDOWNS.with(|countdowns| countdowns[hook as usize].set(nth));
        STATUSES.with(|statuses| statuses[hook as usize].set(status));
    }

    /// Whether this call of `hook` should fail; if so, returns the chosen status at `step`.
    pub(super) fn fails(hook: Hook, step: i32, status: &mut i32) -> Option<i32> {
        let due = COUNTDOWNS.with(|countdowns| {
            let countdown = &countdowns[hook as usize];
            match countdown.get() {
                0 => false,
                1 => {
                    countdown.set(0);
                    true
                }
                left => {
                    countdown.set(left - 1);
                    false
                }
            }
        });
        due.then(|| {
            *status = STATUSES.with(|statuses| statuses[hook as usize].get());
            step
        })
    }
}

/// An import's frame, locked, with the ticket whose commands move it.
///
/// A ticket's command pool, buffers and fences need external synchronization. The lease uses
/// them only through this, which holds the frame's lock (and so the queue lock `FFmpeg` and the
/// renderer share). The pool touches a ticket only before a lease takes it and after the lease
/// gives it back.
struct LeaseAccess<'a> {
    frame: VulkanFrameAccessGuard<'a>,
    ticket: NonNull<c_void>,
}

/// Calls `alpha2_native_gpu_vulkan_acquire_submit`, or fails first where [`inject`] asks.
///
/// # Safety
/// `lease.ticket` was created for the frame's device and queue family, and `lease.frame` may
/// record what the acquire submits.
unsafe fn acquire_submit(lease: &mut LeaseAccess<'_>, status: &mut i32) -> i32 {
    #[cfg(test)]
    if let Some(step) = inject::fails(
        inject::Hook::AcquireSubmit,
        crate::interop::ALPHA2_GPU_STEP_QUEUE_SUBMIT,
        status,
    ) {
        return step;
    }
    // SAFETY: the caller's contract; the guard holds the frame's and the queue's locks.
    unsafe {
        alpha2_native_gpu_vulkan_acquire_submit(
            lease.frame.access_mut(),
            lease.ticket.as_ptr(),
            status,
        )
    }
}

/// Calls `alpha2_native_gpu_vulkan_release`, or fails first where [`inject`] asks.
///
/// # Safety
/// `lease.ticket` is the one the frame's acquire was submitted with, and `lease.frame` may
/// record what the release submits.
unsafe fn release(lease: &mut LeaseAccess<'_>, status: &mut i32) -> i32 {
    #[cfg(test)]
    if let Some(step) = inject::fails(
        inject::Hook::Release,
        crate::interop::ALPHA2_GPU_STEP_QUEUE_SUBMIT,
        status,
    ) {
        return step;
    }
    // SAFETY: the caller's contract; the guard holds the frame's and the queue's locks.
    unsafe {
        alpha2_native_gpu_vulkan_release(lease.frame.access_mut(), lease.ticket.as_ptr(), status)
    }
}

/// Calls `alpha2_native_gpu_vulkan_ticket_poll`, or fails first where [`inject`] asks.
///
/// # Safety
/// `ticket` is a live ticket the caller owns.
unsafe fn ticket_poll(ticket: *const c_void, status: &mut i32) -> i32 {
    #[cfg(test)]
    if let Some(step) = inject::fails(
        inject::Hook::TicketPoll,
        crate::interop::ALPHA2_GPU_STEP_FENCE_STATUS,
        status,
    ) {
        return step;
    }
    // SAFETY: the caller's contract.
    unsafe { alpha2_native_gpu_vulkan_ticket_poll(ticket, status) }
}

/// Calls `alpha2_native_gpu_vulkan_ticket_reset`, or fails first where [`inject`] asks.
///
/// # Safety
/// `ticket` is a live ticket the caller owns, and nothing it submitted is running.
unsafe fn ticket_reset(ticket: *mut c_void, status: &mut i32) -> i32 {
    #[cfg(test)]
    if let Some(step) = inject::fails(
        inject::Hook::TicketReset,
        crate::interop::ALPHA2_GPU_STEP_RESET_FENCE,
        status,
    ) {
        return step;
    }
    // SAFETY: the caller's contract.
    unsafe { alpha2_native_gpu_vulkan_ticket_reset(ticket, status) }
}

use crate::interop::Alpha2VulkanEncodeCopy as VulkanEncodeCopy;

/// Reusable NV12 handoff from the renderer's plane textures to encoder frames, one per export.
///
/// Owns the command pool and command-buffer ring, so a frame costs one queue submission and no
/// CPU wait. The copy is ordered after the render by queue order, not by a semaphore: a
/// submission wgpu rejects signals nothing, and a copy waiting on it would block the shared
/// queue.
pub struct VulkanEncodeCopier {
    copier: *mut c_void,
    /// The lock `FFmpeg`'s submissions and `GpuContext::submit` take on the shared graphics
    /// queue.
    queue_lock: alpha2_ffmpeg_bridge::VulkanQueueLock,
    retained: Option<Box<RetainedEncodeWork>>,
    close_failed: bool,
}

/// Raw Vulkan submissions are invisible to wgpu's tracking, so both ends of every copy and the
/// device stay alive until its fence completes.
struct RetainedEncodeWork {
    copies: Vec<Option<EncodeCopyInputs>>,
    _context: SharedVulkanContext,
    #[cfg(test)]
    lifetime_probe: Option<Arc<()>>,
}

struct EncodeCopyInputs {
    _luma: wgpu::Texture,
    _chroma: wgpu::Texture,
    _frame: rsmpeg::avutil::AVFrame,
}

// SAFETY: the C object is only reached through `&mut self`, and the raw pointer is owned solely by
// this value, which destroys it once.
unsafe impl Send for VulkanEncodeCopier {}

impl fmt::Debug for VulkanEncodeCopier {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("VulkanEncodeCopier")
            .finish_non_exhaustive()
    }
}

impl Drop for VulkanEncodeCopier {
    fn drop(&mut self) {
        if !self.copier.is_null() && !self.close_failed {
            let _ = self.try_close();
        }
        if !self.copier.is_null() {
            // The C helper did not finish, so keep the whole ownership graph alive rather than
            // let Rust free what its commands still use.
            std::mem::forget(self.retained.take());
        }
    }
}

impl VulkanEncodeCopier {
    /// Waits for outstanding copies and releases their resources.
    ///
    /// # Errors
    /// On timeout or device loss, retains the pending GPU resources and device instead of
    /// destroying images raw Vulkan work may still use.
    pub fn finish(mut self) -> Result<(), BridgeError> {
        self.try_close()
    }

    fn try_close(&mut self) -> Result<(), BridgeError> {
        let mut status = 0;
        // SAFETY: created by the matching constructor and destroyed once; the C side waits for
        // in-flight copies before freeing anything.
        let step =
            unsafe { alpha2_native_gpu_vulkan_encode_copier_destroy(self.copier, &raw mut status) };
        if step != 0 {
            self.close_failed = true;
            return Err(helper_error("encode copier teardown", step, status));
        }
        self.copier = std::ptr::null_mut();
        // `retained` is the last owning field, so the device outlives the queue and lock
        // references.
        Ok(())
    }
}

impl VulkanEncodeCopier {
    /// Creates the handoff for one export on `context`'s device.
    ///
    /// # Errors
    /// Returns an error when the command pool, buffers or fences cannot be created, or the
    /// device has no queue lock for its graphics family.
    pub fn new(context: &SharedVulkanContext) -> Result<Self, BridgeError> {
        let gpu = context.gpu();
        // SAFETY: this context is always Vulkan; the handle is only read.
        let device = unsafe { gpu.device().as_hal::<wgpu_hal::api::Vulkan>() }
            .map(|device| device.raw_device().handle().as_raw())
            .ok_or(BridgeError::InvalidInput("renderer is not Vulkan"))?;
        let mut copier = std::ptr::null_mut();
        let mut status = 0;
        // SAFETY: `device` is this renderer's live logical device; the outputs are valid locals.
        let step = unsafe {
            alpha2_native_gpu_vulkan_encode_copier_create(
                device,
                0,
                0,
                &raw mut copier,
                &raw mut status,
            )
        };
        if step != 0 {
            return Err(helper_error("encode copier creation", step, status));
        }
        Ok(Self {
            copier,
            queue_lock: alpha2_ffmpeg_bridge::VulkanQueueLock::new(context.ffmpeg_device(), 0)?,
            retained: Some(Box::new(RetainedEncodeWork {
                copies: (0..Self::slots()).map(|_| None).collect(),
                _context: context.clone(),
                #[cfg(test)]
                lifetime_probe: None,
            })),
            close_failed: false,
        })
    }

    /// Plane-texture pairs a caller must cycle through to feed this copier.
    ///
    /// A single pair is only safe if the caller never runs ahead of the copy. A caller that
    /// encodes on another thread does, so the pairs must be a ring as deep as the copier's
    /// command-buffer ring.
    #[must_use]
    pub fn slots() -> usize {
        // SAFETY: a constant read out of the shim; it touches no state.
        unsafe { alpha2_native_gpu_vulkan_encode_copier_slots() as usize }
    }

    /// Waits until the slot the next copy will use is free, and returns it.
    ///
    /// Call before recording the render that writes the plane textures, and render into this
    /// pair. The wait is on the copy four submissions ago, so a pipeline keeping up never
    /// blocks.
    ///
    /// # Errors
    /// Returns an error when the fence wait fails.
    pub fn await_slot(&mut self) -> Result<usize, BridgeError> {
        let (mut slot, mut status) = (0_u32, 0);
        // SAFETY: the copier is live and the output pointers are valid locals.
        let step = unsafe {
            alpha2_native_gpu_vulkan_encode_copier_await_slot(
                self.copier,
                &raw mut slot,
                &raw mut status,
            )
        };
        if step != 0 {
            return Err(helper_error("encode copy slot", step, status));
        }
        if let Some(retained) = self.retained.as_mut() {
            retained.copies[slot as usize] = None;
        }
        Ok(slot as usize)
    }

    /// Copies a renderer-packed NV12 pair into an encoder frame's image.
    ///
    /// The submission that wrote the textures must already be on the renderer's queue; queue
    /// order alone orders the copy after it. If that submission was rejected, the copy still
    /// runs over whatever the planes held, and the caller fails the frame.
    ///
    /// # Errors
    /// Returns an error when the frame is not one multi-planar image on this device, the
    /// textures are not Vulkan, or the submission fails.
    pub fn submit(
        &mut self,
        frame: &mut alpha2_ffmpeg_bridge::VulkanEncodeFrameAccess<'_>,
        luma: &wgpu::Texture,
        chroma: &wgpu::Texture,
        size: (u32, u32),
    ) -> Result<(), BridgeError> {
        let slot = self.await_slot()?;
        let inputs = EncodeCopyInputs {
            _luma: luma.clone(),
            _chroma: chroma.clone(),
            _frame: frame.retain_frame()?,
        };
        let kept = self
            .retained
            .as_mut()
            .and_then(|work| work.copies.get_mut(slot))
            .ok_or(BridgeError::InvalidInput("invalid encode copier slot"))?;
        // SAFETY: both textures belong to this Vulkan device and are only read; wgpu still owns
        // them.
        let luma_image = unsafe { luma.as_hal::<wgpu_hal::api::Vulkan>() }
            .map(|texture| unsafe { texture.raw_handle() }.as_raw())
            .ok_or(BridgeError::InvalidInput("luma plane is not Vulkan"))?;
        // SAFETY: as above.
        let chroma_image = unsafe { chroma.as_hal::<wgpu_hal::api::Vulkan>() }
            .map(|texture| unsafe { texture.raw_handle() }.as_raw())
            .ok_or(BridgeError::InvalidInput("chroma plane is not Vulkan"))?;
        let copy = VulkanEncodeCopy {
            luma_image,
            chroma_image,
            width: size.0,
            height: size.1,
        };
        // Every submission to the shared graphics queue holds this lock. The plane textures are
        // protected by the caller's slot ring, not queue order, since a caller encoding on
        // another thread can be several frames ahead.
        let queue = self.queue_lock.lock();
        let mut status = 0;
        // SAFETY: the guard holds FFmpeg's frame lock for the call, and the shim validates every
        // handle it reads from the descriptor.
        let step = unsafe {
            alpha2_native_gpu_vulkan_encode_copier_submit(
                self.copier,
                frame.access_mut(),
                &raw const copy,
                &raw mut status,
            )
        };
        drop(queue);
        if step != 0 {
            return Err(helper_error("encode copy", step, status));
        }
        *kept = Some(inputs);
        frame.commit();
        Ok(())
    }
}

/// Checks that a decoded frame can be imported: moves its planes to `SHADER_READ_ONLY` and back
/// in one submission on its queue, through import-like views, and waits.
///
/// # Errors
/// Returns an error when image-view creation, synchronization, or submission fails.
pub fn probe_vulkan_submission(lease: &VulkanFrameLease) -> Result<(), BridgeError> {
    let mut guard = lease.lock_access()?;
    let mut status = 0;
    // SAFETY: the guard holds the FFmpeg frame and queue locks, and the helper uses the
    // descriptor only during this call. It changes only the timeline values, after submitting,
    // and returns the planes to the frame's layouts.
    let step = unsafe { alpha2_native_gpu_vulkan_probe(guard.access_mut(), &raw mut status) };
    // Committed on failure too: a timed-out submission still signals when it runs, and the next
    // user must signal above it.
    guard.commit();
    if step != 0 {
        return Err(helper_error("decode probe", step, status));
    }
    Ok(())
}

/// Coded extents and coding limits one video-encode profile has on a device.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct VideoEncodeExtent {
    /// The profile the codec's Vulkan encoder opens on this device (H.264
    /// `constrained_baseline`, `main` or `high`; HEVC `main`).
    pub profile: &'static str,
    /// Smallest accepted width and height in samples.
    pub minimum: (u32, u32),
    /// Largest accepted width and height in samples.
    pub maximum: (u32, u32),
    /// `VkVideoEncodeCapabilitiesKHR::encodeInputPictureGranularity`, which `FFmpeg` aligns the
    /// coded size up to before the limits apply. So 158x62 can pass a 160x64 minimum while
    /// 64x48 fails; compare aligned sizes, not raw ones.
    pub granularity: (u32, u32),
    /// `maxPPictureL0ReferenceCount`: references a P picture may use.
    pub max_p_picture_l0_references: u32,
    /// `maxBPictureL0ReferenceCount`: forward references a B picture may use. Zero together
    /// with `max_l1_references` means no B pictures.
    pub max_b_picture_l0_references: u32,
    /// `maxL1ReferenceCount`: backward references a B picture may use.
    pub max_l1_references: u32,
    /// `VkVideoEncodeCapabilitiesKHR::rateControlModes` as names: `cbr`, `vbr`, `disabled`
    /// (constant QP).
    pub rate_control_modes: Vec<&'static str>,
}

impl VideoEncodeExtent {
    /// Whether the device can encode B pictures at all.
    #[must_use]
    pub const fn supports_b_pictures(&self) -> bool {
        self.max_b_picture_l0_references > 0 || self.max_l1_references > 0
    }
}

impl VideoEncodeExtent {
    /// Whether an output of this size reaches the encoder's coded limits.
    #[must_use]
    pub const fn accepts(&self, width: u32, height: u32) -> bool {
        let coded = self.coded(width, height);
        coded.0 >= self.minimum.0
            && coded.1 >= self.minimum.1
            && coded.0 <= self.maximum.0
            && coded.1 <= self.maximum.1
    }

    /// The coded extent an output of this size is encoded at.
    #[must_use]
    pub const fn coded(&self, width: u32, height: u32) -> (u32, u32) {
        (
            align_up(width, self.granularity.0),
            align_up(height, self.granularity.1),
        )
    }
}

const fn align_up(value: u32, granularity: u32) -> u32 {
    if granularity <= 1 {
        return value;
    }
    value.div_ceil(granularity).saturating_mul(granularity)
}

/// The H.264 profiles `FFmpeg`'s `h264_vulkan` tries for 8-bit 4:2:0 input, in its order. With
/// no profile requested it opens the last one the device supports (`vulkan_encode_h264.c`,
/// `init_profile`), so `high` where available.
const H264_ENCODE_PROFILES: [(vk::native::StdVideoH264ProfileIdc, &str); 3] = [
    (
        vk::native::StdVideoH264ProfileIdc_STD_VIDEO_H264_PROFILE_IDC_BASELINE,
        "constrained_baseline",
    ),
    (
        vk::native::StdVideoH264ProfileIdc_STD_VIDEO_H264_PROFILE_IDC_MAIN,
        "main",
    ),
    (
        vk::native::StdVideoH264ProfileIdc_STD_VIDEO_H264_PROFILE_IDC_HIGH,
        "high",
    ),
];

/// What `query` answers for the last of `H264_ENCODE_PROFILES` it answers at all: the profile
/// `h264_vulkan` would open, and its answer.
fn last_supported_profile<T>(
    mut query: impl FnMut(vk::native::StdVideoH264ProfileIdc) -> Option<T>,
) -> Option<(&'static str, T)> {
    H264_ENCODE_PROFILES
        .iter()
        .filter_map(|&(idc, name)| query(idc).map(|answer| (name, answer)))
        .last()
}

/// Codecs the shared device encodes where the adapter has the extension: report name,
/// extension, and the operation `FFmpeg` is told the encode queue does. AV1 is omitted because
/// `libsvtav1` encodes it everywhere.
const ENCODE_EXTENSIONS: [(&str, &CStr, vk::VideoCodecOperationFlagsKHR); 2] = [
    (
        "h264",
        ash::khr::video_encode_h264::NAME,
        vk::VideoCodecOperationFlagsKHR::ENCODE_H264,
    ),
    (
        "hevc",
        ash::khr::video_encode_h265::NAME,
        vk::VideoCodecOperationFlagsKHR::ENCODE_H265,
    ),
];

/// The encode operations `codecs` take, for `FFmpeg`'s encode queue.
fn encode_operations(codecs: &[&str]) -> vk::VideoCodecOperationFlagsKHR {
    ENCODE_EXTENSIONS
        .iter()
        .filter(|(codec, _, _)| codecs.contains(codec))
        .fold(
            vk::VideoCodecOperationFlagsKHR::empty(),
            |operations, (_, _, operation)| operations | *operation,
        )
}

/// Queries the coded extents each of `codecs`' 8-bit 4:2:0 encoders accepts, for the profile
/// its Vulkan encoder opens; codecs whose query fails are left out. Asked of the physical
/// device because `FFmpeg` only answers by failing to open a session.
///
/// Only call for a device exposing `VK_KHR_video_queue`, `VK_KHR_video_encode_queue`, and each
/// codec's encode extension.
fn encode_extents(
    entry: &ash::Entry,
    instance: &ash::Instance,
    physical: vk::PhysicalDevice,
    codecs: &[&'static str],
) -> Vec<(&'static str, VideoEncodeExtent)> {
    let video = ash::khr::video_queue::Instance::new(entry, instance);
    codecs
        .iter()
        .filter_map(|&codec| {
            let extent = match codec {
                "h264" => h264_encode_extent(&video, physical),
                "hevc" => hevc_encode_extent(&video, physical),
                _ => None,
            }?;
            Some((codec, extent))
        })
        .collect()
}

/// The 8-bit 4:2:0 capabilities `video` answers for `profile` with `per_codec` chained in, or
/// `None` when the device lacks the profile.
fn encode_capabilities<'a>(
    video: &ash::khr::video_queue::Instance,
    physical: vk::PhysicalDevice,
    profile: &vk::VideoProfileInfoKHR<'_>,
    per_codec: &mut (impl vk::ExtendsVideoCapabilitiesKHR + 'a),
) -> Option<(
    (vk::Extent2D, vk::Extent2D),
    vk::VideoEncodeCapabilitiesKHR<'a>,
)> {
    let mut encode = vk::VideoEncodeCapabilitiesKHR::default();
    let mut capabilities = vk::VideoCapabilitiesKHR::default()
        .push_next(&mut encode)
        .push_next(per_codec);
    // SAFETY: the profile chain lives for the call and the device is this instance's. The caller
    // checked the video queue, encode queue and codec extensions, without which the loader would
    // dispatch to a null entry point; with them, an unsupported profile returns an error.
    let result = unsafe {
        (video.fp().get_physical_device_video_capabilities_khr)(
            physical,
            std::ptr::from_ref(profile),
            &raw mut capabilities,
        )
    };
    let extents = (capabilities.min_coded_extent, capabilities.max_coded_extent);
    (result == vk::Result::SUCCESS).then(|| {
        encode.p_next = std::ptr::null_mut();
        (extents, encode)
    })
}

/// The limits of the H.264 profile `h264_vulkan` opens on the device.
fn h264_encode_extent(
    video: &ash::khr::video_queue::Instance,
    physical: vk::PhysicalDevice,
) -> Option<VideoEncodeExtent> {
    let (profile, mut extent) = last_supported_profile(|idc| {
        let mut h264 = vk::VideoEncodeH264ProfileInfoKHR::default().std_profile_idc(idc);
        let profile = vk::VideoProfileInfoKHR::default()
            .video_codec_operation(vk::VideoCodecOperationFlagsKHR::ENCODE_H264)
            .chroma_subsampling(vk::VideoChromaSubsamplingFlagsKHR::TYPE_420)
            .luma_bit_depth(vk::VideoComponentBitDepthFlagsKHR::TYPE_8)
            .chroma_bit_depth(vk::VideoComponentBitDepthFlagsKHR::TYPE_8)
            .push_next(&mut h264);
        let mut h264_capabilities = vk::VideoEncodeH264CapabilitiesKHR::default();
        let (coded, encode) =
            encode_capabilities(video, physical, &profile, &mut h264_capabilities)?;
        Some(encode_extent(
            coded,
            &encode,
            (
                h264_capabilities.max_p_picture_l0_reference_count,
                h264_capabilities.max_b_picture_l0_reference_count,
                h264_capabilities.max_l1_reference_count,
            ),
        ))
    })?;
    extent.profile = profile;
    Some(extent)
}

/// The limits of HEVC Main, the profile `hevc_vulkan` opens for 8-bit 4:2:0 input.
fn hevc_encode_extent(
    video: &ash::khr::video_queue::Instance,
    physical: vk::PhysicalDevice,
) -> Option<VideoEncodeExtent> {
    let mut hevc = vk::VideoEncodeH265ProfileInfoKHR::default()
        .std_profile_idc(vk::native::StdVideoH265ProfileIdc_STD_VIDEO_H265_PROFILE_IDC_MAIN);
    let profile = vk::VideoProfileInfoKHR::default()
        .video_codec_operation(vk::VideoCodecOperationFlagsKHR::ENCODE_H265)
        .chroma_subsampling(vk::VideoChromaSubsamplingFlagsKHR::TYPE_420)
        .luma_bit_depth(vk::VideoComponentBitDepthFlagsKHR::TYPE_8)
        .chroma_bit_depth(vk::VideoComponentBitDepthFlagsKHR::TYPE_8)
        .push_next(&mut hevc);
    let mut hevc_capabilities = vk::VideoEncodeH265CapabilitiesKHR::default();
    let (coded, encode) = encode_capabilities(video, physical, &profile, &mut hevc_capabilities)?;
    let mut extent = encode_extent(
        coded,
        &encode,
        (
            hevc_capabilities.max_p_picture_l0_reference_count,
            hevc_capabilities.max_b_picture_l0_reference_count,
            hevc_capabilities.max_l1_reference_count,
        ),
    );
    extent.profile = "main";
    Some(extent)
}

/// The limits one profile's capabilities describe; the caller fills in `profile`.
fn encode_extent(
    (minimum, maximum): (vk::Extent2D, vk::Extent2D),
    encode: &vk::VideoEncodeCapabilitiesKHR<'_>,
    (p_l0, b_l0, l1): (u32, u32, u32),
) -> VideoEncodeExtent {
    VideoEncodeExtent {
        profile: "",
        minimum: (minimum.width, minimum.height),
        maximum: (maximum.width, maximum.height),
        // FFmpeg aligns the coded size to this (`vulkan_encode.c`), not to
        // `pictureAccessGranularity`; the two differ on some drivers (32 and 16 on Mesa ANV).
        granularity: (
            encode.encode_input_picture_granularity.width.max(1),
            encode.encode_input_picture_granularity.height.max(1),
        ),
        max_p_picture_l0_references: p_l0,
        max_b_picture_l0_references: b_l0,
        max_l1_references: l1,
        rate_control_modes: [
            (vk::VideoEncodeRateControlModeFlagsKHR::CBR, "cbr"),
            (vk::VideoEncodeRateControlModeFlagsKHR::VBR, "vbr"),
            (vk::VideoEncodeRateControlModeFlagsKHR::DISABLED, "disabled"),
        ]
        .into_iter()
        .filter(|(flag, _)| encode.rate_control_modes.contains(*flag))
        .map(|(_, name)| name)
        .collect(),
    }
}

/// A Vulkan layer's specification version.
#[derive(Copy, Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub struct VulkanLayerVersion {
    /// Major version.
    pub major: u32,
    /// Minor version.
    pub minor: u32,
    /// Patch version (the SDK release).
    pub patch: u32,
}

impl VulkanLayerVersion {
    /// The first Khronos validation layer `FFmpeg`'s video sessions survive. Older ones crash or
    /// hang on the `pNext` chain `FFmpeg` builds for session setup.
    pub const VIDEO_SESSIONS: Self = Self {
        major: 1,
        minor: 4,
        patch: 0,
    };

    /// Unpacks a `VK_MAKE_API_VERSION` value.
    #[must_use]
    pub const fn from_packed(version: u32) -> Self {
        Self {
            major: vk::api_version_major(version),
            minor: vk::api_version_minor(version),
            patch: vk::api_version_patch(version),
        }
    }
}

impl fmt::Display for VulkanLayerVersion {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(formatter, "{}.{}.{}", self.major, self.minor, self.patch)
    }
}

const KHRONOS_VALIDATION: &str = "VK_LAYER_KHRONOS_validation";

/// The version of the Khronos validation layer the loader activated on `instance`, if any.
///
/// Asked of the loader instead of derived from the environment: the deprecated
/// `vkEnumerateDeviceLayerProperties` returns the instance's activated layers, already
/// resolved against `VK_INSTANCE_LAYERS`, `VK_LOADER_LAYERS_ENABLE` and
/// `VK_LOADER_LAYERS_DISABLE` by the loader's own rules, which are easy to reimplement wrong.
fn activated_validation_layer(
    instance: &ash::Instance,
    physical: vk::PhysicalDevice,
) -> Option<VulkanLayerVersion> {
    // SAFETY: `physical` was enumerated from `instance`, which is alive for the call.
    let layers = unsafe { instance.enumerate_device_layer_properties(physical) }.ok()?;
    layers
        .iter()
        .find(|layer| {
            layer
                .layer_name_as_c_str()
                .is_ok_and(|name| name.to_bytes() == KHRONOS_VALIDATION.as_bytes())
        })
        .map(|layer| VulkanLayerVersion::from_packed(layer.spec_version))
}

/// The Vulkan-only HAL instance the shared device is made on, built through `wgpu-hal`
/// directly because it shares no DX12 compiler and has its own flags.
fn hal_instance() -> Result<wgpu_hal::vulkan::Instance, SharedVulkanError> {
    let hal_desc = wgpu_hal::InstanceDescriptor {
        name: "alpha2 shared Vulkan decode",
        flags: wgpu::InstanceFlags::from_env_or_default(),
        memory_budget_thresholds: wgpu::MemoryBudgetThresholds::default(),
        backend_options: wgpu::BackendOptions::from_env_or_default(),
        telemetry: None,
        display: None,
    };
    // SAFETY: the returned HAL instance owns its loader and Vulkan instance.
    unsafe {
        wgpu_hal::vulkan::Instance::init(&hal_desc)
            .map_err(|error| SharedVulkanError(format!("Vulkan instance: {error}")))
    }
}

/// Holds the shared graphics-queue lock for the renderer's submissions.
///
/// Only the [`GpuContext`] it is made for receives it, and that context takes and releases each
/// hold on the same thread, innermost first, within one queue call. That pairing is the raw
/// lock's contract, and no other code can reach this value.
#[derive(Debug)]
struct SharedQueueLock(VulkanQueueLock);

impl ExternalQueueLock for SharedQueueLock {
    fn lock(&self) {
        // SAFETY: the owning context releases each hold with one `unlock` on the same thread, in
        // reverse order.
        unsafe { self.0.raw_lock() };
    }

    fn unlock(&self) {
        // SAFETY: the context releases only its own innermost hold, on the thread that took it.
        unsafe { self.0.raw_unlock() };
    }
}

/// Failure while creating a logical Vulkan device shared by wgpu and `FFmpeg`.
#[derive(Debug)]
pub struct SharedVulkanError(String);

impl fmt::Display for SharedVulkanError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(&self.0)
    }
}

impl Error for SharedVulkanError {}

impl SharedVulkanError {
    /// An error with `message`.
    #[must_use]
    pub fn new(message: impl Into<String>) -> Self {
        Self(message.into())
    }
}

/// An imported bi-planar 4:2:0 image and its two wgpu plane views.
#[derive(Debug)]
pub struct ImportedVulkanFrame {
    /// Plane-zero luma view.
    pub y: wgpu::TextureView,
    /// Plane-one interleaved chroma view.
    pub uv: wgpu::TextureView,
    /// Sample layout of the imported planes.
    pub layout: NativePlaneLayout,
    /// Allocated image extent (the codec's padded coded size, not the display size), which
    /// presentation normalizes sample coordinates against.
    pub coded_size: [u32; 2],
    /// Submission/lifetime lease for the decoder surface.
    pub lease: Arc<VulkanImportLease>,
}

/// Allocation counters of a context's import tickets.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct ImportTicketStats {
    /// Tickets created (a command pool, two command buffers, and a fence each).
    pub created: u64,
    /// Imports served by a finished, reset ticket.
    pub reused: u64,
    /// Tickets reset and ready for an import.
    pub idle: usize,
    /// Tickets whose release may still be running.
    pub pending: usize,
    /// Tickets given up because their device could not report them finished.
    pub abandoned: u64,
}

/// One import ticket and the device and queue family it was made for.
struct PooledTicket {
    ticket: NonNull<c_void>,
    device: u64,
    queue_family: u32,
}

/// A returned ticket waits in `pending` until the fences of what it submitted are seen
/// signalled (polled without blocking at the next import), then is reset and reused. A ticket
/// whose device cannot say, or that is still running when too many wait, is never reused: it
/// is destroyed after a bounded wait or else leaked. Those waits run outside the pool's lock
/// and tickets retired together share one timeout, so a hung device costs it once.
#[derive(Default)]
struct ImportTicketPool {
    state: Mutex<ImportTicketPoolState>,
}

#[derive(Default)]
struct ImportTicketPoolState {
    idle: Vec<PooledTicket>,
    pending: VecDeque<PooledTicket>,
    created: u64,
    reused: u64,
    abandoned: u64,
}

// Frames in flight bound how many tickets an export uses; these bound what a burst leaves.
const MAX_IDLE_IMPORT_TICKETS: usize = 8;
const MAX_PENDING_IMPORT_TICKETS: usize = 32;

// SAFETY: a ticket is touched only by the pool under its mutex, or by the one lease that took
// it, which externally synchronizes its Vulkan objects.
unsafe impl Send for ImportTicketPool {}
// SAFETY: as for `Send`: every access goes through the mutex.
unsafe impl Sync for ImportTicketPool {}

impl ImportTicketPool {
    fn take(
        &self,
        device: u64,
        allocator: u64,
        queue_family: u32,
    ) -> Result<PooledTicket, BridgeError> {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        state.reclaim();
        if let Some(index) = state
            .idle
            .iter()
            .position(|idle| idle.device == device && idle.queue_family == queue_family)
        {
            state.reused = state.reused.saturating_add(1);
            return Ok(state.idle.swap_remove(index));
        }
        state.created = state.created.saturating_add(1);
        drop(state);
        let (mut raw, mut status) = (std::ptr::null_mut(), 0);
        // SAFETY: `device` and `allocator` are the frame's own, read under its lock; the output
        // ticket is owned by the caller.
        let step = unsafe {
            alpha2_native_gpu_vulkan_ticket_create(
                device,
                allocator,
                queue_family,
                &raw mut raw,
                &raw mut status,
            )
        };
        if step != 0 {
            return Err(helper_error("import ticket creation", step, status));
        }
        NonNull::new(raw)
            .map(|ticket| PooledTicket {
                ticket,
                device,
                queue_family,
            })
            .ok_or(BridgeError::InvalidInput(
                "Vulkan import ticket creation returned no ticket",
            ))
    }

    /// Takes back a ticket whatever it submitted; each submission has a fence the pool polls
    /// before reuse.
    fn give_back(&self, ticket: PooledTicket) {
        let mut state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        state.pending.push_back(ticket);
        if state.pending.len() > MAX_PENDING_IMPORT_TICKETS {
            state.reclaim();
        }
        let excess = state
            .pending
            .len()
            .saturating_sub(MAX_PENDING_IMPORT_TICKETS);
        let oldest = state.pending.drain(..excess).collect::<Vec<_>>();
        state.abandoned = state.abandoned.saturating_add(oldest.len() as u64);
        // Imports of the device go on while these wait.
        drop(state);
        retire(oldest);
    }

    fn stats(&self) -> ImportTicketStats {
        let state = self
            .state
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        ImportTicketStats {
            created: state.created,
            reused: state.reused,
            idle: state.idle.len(),
            pending: state.pending.len(),
            abandoned: state.abandoned,
        }
    }
}

impl ImportTicketPoolState {
    /// Resets every pending ticket whose release has finished and makes it idle, without
    /// waiting.
    fn reclaim(&mut self) {
        let mut index = 0;
        while index < self.pending.len() {
            let ticket = self.pending[index].ticket;
            let mut status = 0;
            // SAFETY: the pool owns every pending ticket.
            match unsafe { ticket_poll(ticket.as_ptr(), &mut status) } {
                0 => index += 1,
                1 => {
                    let Some(finished) = self.pending.remove(index) else {
                        break;
                    };
                    // SAFETY: nothing the ticket submitted is still running.
                    let reset = unsafe { ticket_reset(ticket.as_ptr(), &mut status) };
                    if reset != 0 {
                        self.abandoned = self.abandoned.saturating_add(1);
                    } else if self.idle.len() < MAX_IDLE_IMPORT_TICKETS {
                        self.idle.push(finished);
                    } else {
                        // SAFETY: finished and reset, so nothing uses it.
                        unsafe { alpha2_native_gpu_vulkan_import_destroy(ticket.as_ptr()) };
                    }
                }
                _ => {
                    // The device cannot say whether it finished, so leak it rather than free it
                    // while running.
                    self.pending.remove(index);
                    self.abandoned = self.abandoned.saturating_add(1);
                }
            }
        }
    }
}

impl Drop for ImportTicketPool {
    fn drop(&mut self) {
        let state = self
            .state
            .get_mut()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        for idle in state.idle.drain(..) {
            // SAFETY: idle tickets are reset and unused.
            unsafe { alpha2_native_gpu_vulkan_import_destroy(idle.ticket.as_ptr()) };
        }
        retire(state.pending.drain(..));
    }
}

/// Destroys owned tickets once their work has finished, all within one wait timeout; a ticket
/// still unfinished then is leaked rather than freed under the GPU.
fn retire(tickets: impl IntoIterator<Item = PooledTicket>) {
    #[cfg(test)]
    inject::retire_begins();
    let deadline = std::time::Instant::now().checked_add(wait_timeout());
    for pooled in tickets {
        let left = deadline.map_or(std::time::Duration::MAX, |deadline| {
            deadline.saturating_duration_since(std::time::Instant::now())
        });
        let nanoseconds = u64::try_from(left.as_nanos()).unwrap_or(u64::MAX);
        // SAFETY: the pool gives up the ticket here; the helper frees it only after its fences
        // signal.
        unsafe { alpha2_native_gpu_vulkan_import_retire(pooled.ticket.as_ptr(), nanoseconds) };
    }
}

/// Owns the imported texture, `FFmpeg` frame, and native transition commands.
pub struct VulkanImportLease {
    frame: Arc<VulkanFrameLease>,
    _texture: wgpu::Texture,
    gpu: GpuContext,
    batch: Arc<VulkanSubmissionBatch>,
    tickets: Arc<ImportTicketPool>,
    /// Taken only by the drop, which returns it to `tickets`; used only through [`Self::lock`].
    ticket: Option<PooledTicket>,
    staged_signals: Mutex<Vec<StagedSignal>>,
    queued: AtomicBool,
    acquired: AtomicBool,
    released: AtomicBool,
    retained_bytes: u64,
}

#[derive(Clone, Copy, Debug)]
struct StagedSignal {
    semaphore: u64,
    acquired_value: u64,
    signal_value: u64,
}

#[derive(Default)]
struct VulkanSubmissionBatch {
    pending: Mutex<Vec<Weak<VulkanImportLease>>>,
}

impl VulkanSubmissionBatch {
    fn enqueue(&self, lease: &Arc<VulkanImportLease>) {
        if !lease.queued.swap(true, Ordering::AcqRel) {
            self.pending
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .push(Arc::downgrade(lease));
        }
    }

    fn flush(&self) -> Result<(), BridgeError> {
        let leases = self
            .pending
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .drain(..)
            .filter_map(|lease| lease.upgrade())
            .filter(|lease| !lease.acquired.load(Ordering::Acquire))
            .collect::<Vec<_>>();
        if leases.is_empty() {
            return Ok(());
        }
        // Draining cleared the queue, so leases still unacquired after a failure must be
        // re-enqueueable.
        let result = flush_leases(&leases);
        if result.is_err() {
            for lease in &leases {
                if !lease.acquired.load(Ordering::Acquire) {
                    lease.queued.store(false, Ordering::Release);
                }
            }
        }
        result
    }
}

/// Acquires every queued import for the renderer's next submission.
///
/// Each frame is acquired under its own lock from its current state: the barrier starts from
/// its current layout and waits on the value `FFmpeg` last signalled. Batching all barriers
/// under one lock would let an intervening decode (using a frame as a reference) move a frame
/// on, so the batch would wait on a stale value.
///
/// The sampling submission's signal is staged under that same lock, before `FFmpeg` sees the
/// frame again. Otherwise a decode referencing the frame could move it to its DPB layout while
/// the renderer still samples it (VUID-vkCmdDraw-None-09600) and signal the frame's timeline
/// from the decode queue out of order with the renderer, leaving `FFmpeg` waiting forever.
///
/// Only one frame lock is held at a time, so this cannot deadlock against a decode that locks
/// several frames; the cost is one queue submission per imported frame.
fn flush_leases(leases: &[Arc<VulkanImportLease>]) -> Result<(), BridgeError> {
    for lease in leases {
        let mut access = lease.lock()?;
        // Decided under the frame's lock, so a lease two flushes reach is acquired once.
        if lease.acquired.load(Ordering::Acquire) {
            continue;
        }
        let mut status = 0;
        // SAFETY: the ticket was created for this retained frame at import, and the frame and queue
        // stay locked for the submission.
        let step = unsafe { acquire_submit(&mut access, &mut status) };
        if step != 0 {
            return Err(helper_error("import acquire", step, status));
        }
        lease.acquired.store(true, Ordering::Release);
        // The acquire is submitted regardless of staging, so commit its state now.
        access.frame.commit();
        // SAFETY: staging records as the frame's timeline values the signals added to the
        // renderer's next submission, which samples this frame; the layouts stay those the
        // acquire recorded.
        lease.stage_submission_signal(unsafe { access.frame.access_mut() })?;
    }
    Ok(())
}

// SAFETY: shared state is behind locks. The ticket's Vulkan objects are used only through
// `LeaseAccess`, under the frame's lock (which also holds the shared queue lock), and the drop
// has the lease to itself. Staged signals are behind a mutex, flags are atomics, and the frame
// and wgpu objects synchronize themselves.
unsafe impl Send for VulkanImportLease {}
// SAFETY: as for `Send`: `&self` reaches the ticket only through `Self::lock`.
unsafe impl Sync for VulkanImportLease {}

impl fmt::Debug for VulkanImportLease {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("VulkanImportLease")
            .field("released", &self.released.load(Ordering::Acquire))
            .field("retained_bytes", &self.retained_bytes)
            .finish_non_exhaustive()
    }
}

impl VulkanImportLease {
    /// Locks the frame for a use of the ticket, the only way `&self` reaches it.
    fn lock(&self) -> Result<LeaseAccess<'_>, BridgeError> {
        let ticket = self
            .ticket
            .as_ref()
            .map(|pooled| pooled.ticket)
            .ok_or(BridgeError::InvalidInput("missing Vulkan import ticket"))?;
        Ok(LeaseAccess {
            frame: self.frame.lock_access()?,
            ticket,
        })
    }

    /// Approximate decoded-surface bytes retained by this import.
    #[must_use]
    pub const fn retained_bytes(&self) -> u64 {
        self.retained_bytes
    }

    /// Timeline signals this lease has staged on the wgpu queue: nonzero only between staging
    /// the sampling submission and its completion. A nonzero count when the next frame starts
    /// staging means a stale signal would ride on it.
    #[must_use]
    pub fn staged_signals(&self) -> usize {
        self.staged_signals
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .len()
    }

    /// Queues this import for the next [`Self::flush_submission`], which acquires the frame and
    /// adds its timeline signals to the sampling wgpu submission. Queuing twice is harmless.
    pub fn prepare_submission(self: &Arc<Self>) {
        self.batch.enqueue(self);
    }

    /// Acquires every queued import of this context, one native submission per frame under that
    /// frame's lock, and stages the signals of the sampling wgpu submission.
    ///
    /// # Errors
    /// Returns an error if a frame cannot be locked, its acquire fails, or its signals cannot be
    /// staged. Imports not yet acquired can be queued again for a retry.
    pub fn flush_submission(&self) -> Result<(), BridgeError> {
        self.batch.flush()
    }

    /// Stages the sampling submission's timeline signals and records them as the frame's values.
    /// `access` is the frame's lock, held since the acquire.
    fn stage_submission_signal(&self, access: &mut VulkanFrameAccess) -> Result<(), BridgeError> {
        let mut staged = self
            .staged_signals
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if !staged.is_empty() || self.released.load(Ordering::Acquire) {
            return Ok(());
        }
        // SAFETY: the queue was created from this context's Vulkan OpenDevice.
        let queue = unsafe { self.gpu.queue().as_hal::<wgpu_hal::api::Vulkan>() }
            .ok_or(BridgeError::InvalidInput("renderer queue is not Vulkan"))?;
        let descriptor = access.frame_mut();
        for index in 0..descriptor.images.len() {
            if descriptor.images[index] == 0 {
                continue;
            }
            let semaphore = descriptor.semaphores[index];
            let acquired_value = descriptor.semaphore_values[index];
            if semaphore == 0 {
                return Err(BridgeError::InvalidInput("frame has no timeline semaphore"));
            }
            if let Some(existing) = staged
                .iter_mut()
                .find(|existing| existing.semaphore == semaphore)
            {
                existing.acquired_value = existing.acquired_value.max(acquired_value);
                existing.signal_value = existing.acquired_value.saturating_add(1);
            } else {
                staged.push(StagedSignal {
                    semaphore,
                    acquired_value,
                    signal_value: acquired_value.saturating_add(1),
                });
            }
        }
        for signal in &*staged {
            queue.add_signal_semaphore(
                vk::Semaphore::from_raw(signal.semaphore),
                Some(signal.signal_value),
            );
            for index in 0..descriptor.images.len() {
                if descriptor.semaphores[index] == signal.semaphore {
                    descriptor.semaphore_values[index] = signal.signal_value;
                }
            }
        }
        Ok(())
    }

    /// Records that the wgpu submission sampling this import was made.
    ///
    /// If the frame's signals were staged on it, the frame already records its value and
    /// layout. Otherwise the planes are returned to `FFmpeg`'s layout in a submission of their
    /// own.
    ///
    /// # Errors
    /// Returns an error if the frame can no longer be locked or submitted.
    pub fn submitted(&self) -> Result<(), BridgeError> {
        if self.released.load(Ordering::Acquire) {
            return Ok(());
        }
        if !self
            .staged_signals
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .is_empty()
        {
            self.released.store(true, Ordering::Release);
            return Ok(());
        }
        let mut access = self.lock()?;
        // Decided under the frame's lock, so a lease two callers reach is released once.
        if self.released.load(Ordering::Acquire) {
            return Ok(());
        }
        let mut status = 0;
        // SAFETY: the ticket was created for this retained frame and this lease owns it until
        // queue completion.
        let step = unsafe { release(&mut access, &mut status) };
        if step != 0 {
            return Err(helper_error("import release", step, status));
        }
        access.frame.commit();
        self.released.store(true, Ordering::Release);
        Ok(())
    }
}

impl Drop for VulkanImportLease {
    /// Leaves the frame in a state `FFmpeg` can continue from, and returns the ticket.
    ///
    /// An acquired lease whose sampling submission never released it still owes the frame its
    /// signal. If the signal was staged, the frame records that value and a later decode may
    /// already wait on it, so removing the signal alone would hang the decode queue. Under the
    /// frame's lock (held by `FFmpeg` from reading the value to submitting its wait), an
    /// unsubmitted signal is taken off wgpu's queue and raised by a submission of its own after
    /// the acquire. A lease with nothing staged releases the frame from its current state.
    ///
    /// The ticket goes back to the pool, which reuses it only after every fence it submitted
    /// has signalled, since an acquire may still be waiting on the decoder.
    fn drop(&mut self) {
        let Some(pooled) = self.ticket.take() else {
            return;
        };
        if self.acquired.load(Ordering::Acquire) && !self.released.load(Ordering::Acquire) {
            let staged = std::mem::take(
                self.staged_signals
                    .get_mut()
                    .unwrap_or_else(std::sync::PoisonError::into_inner),
            );
            if let Ok(frame) = self.frame.lock_access() {
                // The drop has the lease, and so its ticket, to itself.
                let mut lease = LeaseAccess {
                    frame,
                    ticket: pooled.ticket,
                };
                if staged.is_empty() {
                    let mut status = 0;
                    // SAFETY: the ticket is this frame's and the guard holds its lock.
                    if unsafe { release(&mut lease, &mut status) } == 0 {
                        lease.frame.commit();
                    }
                } else {
                    // SAFETY: the queue belongs to the shared Vulkan device.
                    let queue = unsafe { self.gpu.queue().as_hal::<wgpu_hal::api::Vulkan>() };
                    for signal in &staged {
                        let unsubmitted = queue.as_ref().is_some_and(|queue| {
                            queue.remove_signal_semaphore(vk::Semaphore::from_raw(signal.semaphore))
                        });
                        if unsubmitted {
                            let mut status = 0;
                            // SAFETY: the guard holds the frame's and queue's locks; the
                            // semaphore is the frame's timeline, which the acquire takes to
                            // `acquired_value`.
                            let _ = unsafe {
                                alpha2_native_gpu_vulkan_signal(
                                    lease.frame.access_mut(),
                                    signal.semaphore,
                                    signal.acquired_value,
                                    signal.signal_value,
                                    &raw mut status,
                                )
                            };
                        }
                    }
                }
            }
        }
        self.tickets.give_back(pooled);
    }
}

/// Device features `FFmpeg`'s Vulkan code uses that wgpu does not enable.
///
/// `FFmpeg` enables these on devices it creates, and uses them regardless on this wgpu-created
/// one: `vkQueueSubmit2` needs `synchronization2`, YCbCr conversion needs
/// `samplerYcbcrConversion`, profile-independent images need `videoMaintenance1`, and VP9
/// decode needs `videoDecodeVP9`.
#[derive(Clone, Copy, Debug, Default)]
#[allow(clippy::struct_excessive_bools)]
struct FfmpegDeviceFeatures {
    synchronization2: bool,
    sampler_ycbcr_conversion: bool,
    video_maintenance1: bool,
    video_decode_vp9: bool,
}

/// The [`FfmpegDeviceFeatures`] `physical` supports; `video_maintenance1` and `video_decode_vp9`
/// are only asked for when their extensions are enabled.
fn ffmpeg_device_features(
    instance: &ash::Instance,
    physical: vk::PhysicalDevice,
    video_maintenance1: bool,
    video_decode_vp9: bool,
) -> FfmpegDeviceFeatures {
    let mut synchronization2 = vk::PhysicalDeviceSynchronization2Features::default();
    let mut ycbcr = vk::PhysicalDeviceSamplerYcbcrConversionFeatures::default();
    let mut maintenance = vk::PhysicalDeviceVideoMaintenance1FeaturesKHR::default();
    let mut vp9 = VideoDecodeVp9Features::default();
    let mut features = vk::PhysicalDeviceFeatures2::default()
        .push_next(&mut synchronization2)
        .push_next(&mut ycbcr);
    if video_maintenance1 {
        features = features.push_next(&mut maintenance);
    }
    if video_decode_vp9 {
        features = features.push_next(&mut vp9);
    }
    // SAFETY: `physical` was enumerated from `instance`, and every chained structure outlives
    // the call.
    unsafe { instance.get_physical_device_features2(physical, &mut features) };
    FfmpegDeviceFeatures {
        synchronization2: synchronization2.synchronization2 == vk::TRUE,
        sampler_ycbcr_conversion: ycbcr.sampler_ycbcr_conversion == vk::TRUE,
        video_maintenance1: video_maintenance1 && maintenance.video_maintenance1 == vk::TRUE,
        video_decode_vp9: video_decode_vp9 && vp9.video_decode_vp9 == vk::TRUE,
    }
}

/// `VK_KHR_video_decode_vp9`, which ash 0.38 (Vulkan 1.3.281) predates. Values from
/// `vulkan_core.h` 1.4.330.
const VIDEO_DECODE_VP9_NAME: &CStr = c"VK_KHR_video_decode_vp9";
const DECODE_VP9: vk::VideoCodecOperationFlagsKHR = vk::VideoCodecOperationFlagsKHR::from_raw(0x8);
const VIDEO_DECODE_VP9_FEATURES: vk::StructureType = vk::StructureType::from_raw(1_000_514_000);

/// `VkPhysicalDeviceVideoDecodeVP9FeaturesKHR` (`vulkan_core.h` 1.4.330, line 13215).
#[repr(C)]
#[derive(Clone, Copy, Debug)]
struct VideoDecodeVp9Features {
    s_type: vk::StructureType,
    p_next: *mut c_void,
    video_decode_vp9: vk::Bool32,
}

impl Default for VideoDecodeVp9Features {
    fn default() -> Self {
        Self {
            s_type: VIDEO_DECODE_VP9_FEATURES,
            p_next: std::ptr::null_mut(),
            video_decode_vp9: vk::FALSE,
        }
    }
}

// SAFETY: this is the C structure, starting with `sType` and `pNext`, and the specification
// lists it as extending both.
unsafe impl vk::ExtendsPhysicalDeviceFeatures2 for VideoDecodeVp9Features {}
// SAFETY: as above.
unsafe impl vk::ExtendsDeviceCreateInfo for VideoDecodeVp9Features {}

impl FfmpegDeviceFeatures {
    /// Appends the supported features to the device's creation chain. wgpu builds its own
    /// feature structures afterwards and uses none of these. They live for the process because
    /// the chain is read after the callback returns.
    fn enable(self, create_info: &mut vk::DeviceCreateInfo<'_>) {
        let mut info = std::mem::take(create_info);
        if self.synchronization2 {
            info = info.push_next(Box::leak(Box::new(
                vk::PhysicalDeviceSynchronization2Features::default().synchronization2(true),
            )));
        }
        if self.sampler_ycbcr_conversion {
            info = info.push_next(Box::leak(Box::new(
                vk::PhysicalDeviceSamplerYcbcrConversionFeatures::default()
                    .sampler_ycbcr_conversion(true),
            )));
        }
        if self.video_maintenance1 {
            info = info.push_next(Box::leak(Box::new(
                vk::PhysicalDeviceVideoMaintenance1FeaturesKHR::default().video_maintenance1(true),
            )));
        }
        if self.video_decode_vp9 {
            info = info.push_next(Box::leak(Box::new(VideoDecodeVp9Features {
                video_decode_vp9: vk::TRUE,
                ..VideoDecodeVp9Features::default()
            })));
        }
        *create_info = info;
    }
}

/// One Vulkan logical device adopted by both the renderer and `FFmpeg`.
#[derive(Clone)]
pub struct SharedVulkanContext {
    // Fields drop in order; idle import tickets are Vulkan objects of the device, so they go
    // first.
    import_tickets: Arc<ImportTicketPool>,
    // Drop FFmpeg references before wgpu destroys the raw device.
    ffmpeg_device: AVHWDeviceContext,
    gpu: GpuContext,
    submission_batch: Arc<VulkanSubmissionBatch>,
    identity: NativeGpuIdentity,
    decode_codecs: Vec<&'static str>,
    encode_codecs: Vec<&'static str>,
    encode_extents: Vec<(&'static str, VideoEncodeExtent)>,
    validation_layer: Option<VulkanLayerVersion>,
}

impl fmt::Debug for SharedVulkanContext {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("SharedVulkanContext")
            .field("identity", &self.identity)
            .field("decode_codecs", &self.decode_codecs)
            .field("encode_codecs", &self.encode_codecs)
            .finish_non_exhaustive()
    }
}

impl SharedVulkanContext {
    /// Creates the shared device on the adapter the host's choice makes by default
    /// ([`choose_shared_vulkan`] under `ALPHA2_GPU_ADAPTER`, decode allowed).
    ///
    /// # Errors
    /// Returns why no adapter carries the shared device.
    pub fn create() -> Result<Self, SharedVulkanError> {
        let choice = choose_shared_vulkan(&alpha2_gpu::AdapterPolicy::from_env(), true, true)?;
        choice.context.ok_or_else(|| {
            SharedVulkanError(
                choice
                    .unavailable
                    .unwrap_or_else(|| choice.report.explanation.clone()),
            )
        })
    }

    /// Creates the shared Vulkan 1.3 device, with wgpu, video decode, and NV12 support, on the
    /// adapter at Vulkan enumeration `index` (as `ALPHA2_GPU_ADAPTER` names it).
    ///
    /// # Errors
    /// Returns why the adapter cannot carry the shared device, or the failing initialization
    /// step.
    #[allow(clippy::too_many_lines)]
    pub fn create_on(index: usize) -> Result<Self, SharedVulkanError> {
        let _turn = alpha2_gpu::serialize_acquisition();
        let hal_instance = hal_instance()?;
        // SAFETY: adapter enumeration does not require a surface.
        let mut adapters = unsafe { hal_instance.enumerate_adapters(None) };
        if index >= adapters.len() {
            return Err(SharedVulkanError(format!(
                "there is no Vulkan adapter {index}; there are {}",
                adapters.len()
            )));
        }
        let exposed = adapters.swap_remove(index);
        let physical = exposed.adapter.raw_physical_device();
        let shared_instance = exposed.adapter.shared_instance();
        let raw_instance = shared_instance.raw_instance().handle();
        let validation_layer = activated_validation_layer(shared_instance.raw_instance(), physical);
        let VideoPlan {
            decode_family,
            encode_family,
            mut video_extensions,
            mut decode_caps,
            mut decode_codecs,
            encode_codecs,
        } = adapter_video_plan(shared_instance.raw_instance(), &exposed)
            .map_err(SharedVulkanError)?;
        let graphics_family = 0;
        // Queried only once the device is known to expose the video queue and codec encode
        // extensions (otherwise the loader trampoline has no function behind it), and before
        // the adapter moves into wgpu.
        let encode_extents = if encode_family.is_some() {
            encode_extents(
                shared_instance.entry(),
                shared_instance.raw_instance(),
                physical,
                &encode_codecs,
            )
        } else {
            Vec::new()
        };
        // P010 is optional: without it 8-bit streams still decode and 10-bit sources fall back
        // to software.
        let required_features = (exposed.features
            & (wgpu::Features::TIMESTAMP_QUERY
                | wgpu::Features::TEXTURE_FORMAT_P010
                | wgpu::Features::TEXTURE_FORMAT_16BIT_NORM))
            | wgpu::Features::TEXTURE_FORMAT_NV12;
        let required_limits = wgpu::Limits::downlevel_webgl2_defaults()
            .using_resolution(exposed.capabilities.limits.clone());
        let enabled_extensions = exposed
            .adapter
            .required_device_extensions(required_features);
        let core = exposed
            .adapter
            .physical_device_features(&enabled_extensions, required_features)
            .get_core();
        let ffmpeg_features = ffmpeg_device_features(
            shared_instance.raw_instance(),
            physical,
            video_extensions.contains(&ash::khr::video_maintenance1::NAME),
            video_extensions.contains(&VIDEO_DECODE_VP9_NAME),
        );
        // A VP9 session also needs the extension's feature; without it VP9 decodes in software.
        if !ffmpeg_features.video_decode_vp9 {
            video_extensions.retain(|extension| *extension != VIDEO_DECODE_VP9_NAME);
            decode_caps &= !DECODE_VP9;
            decode_codecs.retain(|codec| *codec != "vp9");
        }
        let extra_family = decode_family;
        let extra_encode_family =
            encode_family.filter(|family| *family != graphics_family && *family != decode_family);
        let extra_extensions = video_extensions.clone();
        // SAFETY: all added extensions and queues were queried from this adapter.
        let open = unsafe {
            exposed.adapter.open_with_callback(
                required_features,
                &required_limits,
                &wgpu::MemoryHints::Performance,
                Some(Box::new(move |args| {
                    ffmpeg_features.enable(args.create_info);
                    for extension in extra_extensions {
                        if !args.extensions.contains(&extension) {
                            args.extensions.push(extension);
                        }
                    }
                    if extra_family != 0 {
                        args.queue_create_infos.push(
                            vk::DeviceQueueCreateInfo::default()
                                .queue_family_index(extra_family)
                                .queue_priorities(&[1.0]),
                        );
                    }
                    if let Some(family) = extra_encode_family {
                        args.queue_create_infos.push(
                            vk::DeviceQueueCreateInfo::default()
                                .queue_family_index(family)
                                .queue_priorities(&[1.0]),
                        );
                    }
                })),
            )
        }
        .map_err(|error| SharedVulkanError(format!("shared Vulkan device: {error}")))?;
        let raw_device = open.device.raw_device().handle();
        let device_extensions = open
            .device
            .enabled_device_extensions()
            .iter()
            .map(|name| name.to_string_lossy().into_owned())
            .collect::<Vec<_>>();
        let instance_extensions = shared_instance
            .extensions()
            .iter()
            .map(|name| name.to_string_lossy().into_owned())
            .collect::<Vec<_>>();
        // SAFETY: the instance, physical device, and device are the ones wgpu just opened with
        // these queues and extensions, and `Self` drops the `FFmpeg` context first.
        let ffmpeg_device = unsafe {
            create_vulkan_device_context(&VulkanDeviceInfo {
                instance: raw_instance.as_raw(),
                physical_device: physical.as_raw(),
                device: raw_device.as_raw(),
                graphics_queue_family: 0,
                graphics_queue_count: 1,
                decode_queue_family: decode_family,
                decode_queue_count: 1,
                decode_video_caps: decode_caps.as_raw(),
                encode_queue_family: encode_family.unwrap_or(0),
                encode_queue_count: u32::from(encode_family.is_some()),
                encode_video_caps: encode_family
                    .map_or(0, |_| encode_operations(&encode_codecs).as_raw()),
                core_features: [
                    core.shader_image_gather_extended,
                    core.fragment_stores_and_atomics,
                    core.shader_int64,
                    core.vertex_pipeline_stores_and_atomics,
                ],
                synchronization2: u32::from(ffmpeg_features.synchronization2),
                sampler_ycbcr_conversion: u32::from(ffmpeg_features.sampler_ycbcr_conversion),
                video_maintenance1: u32::from(ffmpeg_features.video_maintenance1),
                video_decode_vp9: u32::from(ffmpeg_features.video_decode_vp9),
                instance_extensions,
                device_extensions,
            })
        }
        .map_err(|error| SharedVulkanError(format!("FFmpeg Vulkan device: {error}")))?;
        // SAFETY: the adapter and instance originate from the same HAL object.
        let wgpu_instance =
            unsafe { wgpu::Instance::from_hal::<wgpu_hal::api::Vulkan>(hal_instance) };
        // SAFETY: `exposed` was enumerated from the HAL instance `wgpu_instance` now wraps.
        let adapter = unsafe { wgpu_instance.create_adapter_from_hal(exposed) };
        let descriptor = wgpu::DeviceDescriptor {
            label: Some("alpha2 shared Vulkan decode"),
            required_features,
            required_limits,
            ..Default::default()
        };
        // SAFETY: OpenDevice has the descriptor's exact features and limits.
        let (device, queue) = unsafe { adapter.create_device_from_hal(open, &descriptor) }
            .map_err(|error| SharedVulkanError(format!("wgpu adoption: {error}")))?;
        // wgpu and FFmpeg submit to the same graphics queue, which Vulkan requires to be
        // externally synchronized, so both take the same lock.
        let queue_lock: Arc<dyn ExternalQueueLock> = Arc::new(SharedQueueLock(
            VulkanQueueLock::new(ffmpeg_device.clone(), 0)
                .map_err(|error| SharedVulkanError(format!("FFmpeg Vulkan queue lock: {error}")))?,
        ));
        let gpu = GpuContext::from_parts_with_queue_lock(adapter, device, queue, Some(queue_lock));
        Ok(Self {
            ffmpeg_device,
            gpu,
            submission_batch: Arc::new(VulkanSubmissionBatch::default()),
            import_tickets: Arc::new(ImportTicketPool::default()),
            identity: NativeGpuIdentity {
                physical_device: physical.as_raw(),
                device: raw_device.as_raw(),
                epoch: 1,
            },
            decode_codecs,
            encode_extents,
            encode_codecs,
            validation_layer,
        })
    }

    /// Renderer context backed by the shared logical device.
    #[must_use]
    pub fn gpu(&self) -> GpuContext {
        self.gpu.clone()
    }

    /// `FFmpeg` device reference for one decoder session.
    #[must_use]
    pub fn ffmpeg_device(&self) -> AVHWDeviceContext {
        self.ffmpeg_device.clone()
    }

    /// Raw-device cache identity.
    #[must_use]
    pub const fn identity(&self) -> NativeGpuIdentity {
        self.identity
    }

    /// Whether this device can present 10-bit decoded planes.
    #[must_use]
    pub fn supports_ten_bit(&self) -> bool {
        self.gpu
            .device()
            .features()
            .contains(wgpu::Features::TEXTURE_FORMAT_P010)
    }

    /// Decode codecs (`h264`, `hevc`, `av1`, `vp9`) whose extensions the device
    /// was created with.
    #[must_use]
    pub fn decode_codecs(&self) -> &[&'static str] {
        &self.decode_codecs
    }

    /// The Khronos validation layer the loader activated on this instance (by wgpu's
    /// validation flag or the environment), and its version.
    #[must_use]
    pub const fn validation_layer(&self) -> Option<VulkanLayerVersion> {
        self.validation_layer
    }

    /// Why `FFmpeg` must not open a video session on this device, or `None` when it may.
    ///
    /// Validation layers older than [`VulkanLayerVersion::VIDEO_SESSIONS`] crash or hang in
    /// session setup, for decode and encode alike, so both paths ask this.
    #[must_use]
    pub fn video_session_hazard(&self) -> Option<String> {
        self.validation_layer
            .filter(|layer| *layer < VulkanLayerVersion::VIDEO_SESSIONS)
            .map(|layer| {
                format!(
                    "the loaded Vulkan validation layer is {layer}, and layers older than {} \
                     crash in FFmpeg's video session setup",
                    VulkanLayerVersion::VIDEO_SESSIONS
                )
            })
    }

    /// Video-encode codecs this device created a queue and extensions for. Empty without an
    /// encode queue, in which case `FFmpeg`'s Vulkan encoder cannot open.
    #[must_use]
    pub fn encode_codecs(&self) -> &[&'static str] {
        &self.encode_codecs
    }

    /// Output sizes this device's encoder of `codec` (`h264`, `hevc`) accepts, from
    /// `vkGetPhysicalDeviceVideoCapabilitiesKHR`. Exports outside it take the readback path.
    ///
    /// `None` when the device does not encode `codec` or the query failed.
    #[must_use]
    pub fn encode_extent(&self, codec: &str) -> Option<&VideoEncodeExtent> {
        self.encode_extents
            .iter()
            .find(|(encoded, _)| *encoded == codec)
            .map(|(_, extent)| extent)
    }

    /// Allocation counters of the import tickets this device reuses.
    #[must_use]
    pub fn import_ticket_stats(&self) -> ImportTicketStats {
        self.import_tickets.stats()
    }

    /// Imports one same-device `FFmpeg` NV12 or P010 surface as plane views.
    ///
    /// The acquire waits on `FFmpeg`'s latest timeline value without a CPU wait. Call
    /// `submitted()` on the returned lease right after the wgpu submission that samples it.
    ///
    /// # Errors
    /// Returns an error for a different device, an unsupported image layout, or a failed Vulkan
    /// transition submission.
    #[allow(clippy::too_many_lines)]
    pub fn import_surface(
        &self,
        frame: Arc<VulkanFrameLease>,
    ) -> Result<ImportedVulkanFrame, BridgeError> {
        let info = frame.info()?;
        if info.device != self.identity.device
            || info.physical_device != self.identity.physical_device
        {
            return Err(BridgeError::InvalidInput(
                "decoded frame belongs to another Vulkan device",
            ));
        }
        let layout = match (info.software_format, info.formats[0]) {
            (rsmpeg::ffi::AV_PIX_FMT_NV12, format)
                if format
                    == vk::Format::G8_B8R8_2PLANE_420_UNORM
                        .as_raw()
                        .cast_unsigned() =>
            {
                NativePlaneLayout::Nv12
            }
            (rsmpeg::ffi::AV_PIX_FMT_P010LE, format)
                if format
                    == vk::Format::G10X6_B10X6R10X6_2PLANE_420_UNORM_3PACK16
                        .as_raw()
                        .cast_unsigned() =>
            {
                NativePlaneLayout::P010
            }
            _ => {
                return Err(BridgeError::InvalidInput(
                    "only sampled single-image NV12 or P010 is importable",
                ));
            }
        };
        if layout == NativePlaneLayout::P010
            && !self
                .gpu
                .device()
                .features()
                .contains(wgpu::Features::TEXTURE_FORMAT_P010)
        {
            return Err(BridgeError::InvalidInput(
                "shared Vulkan device does not expose P010 plane views",
            ));
        }
        if info.plane_count != 1 || !info.sampled || info.width <= 0 || info.height <= 0 {
            return Err(BridgeError::InvalidInput(
                "only sampled single-image NV12 or P010 is importable",
            ));
        }
        // The image's extent is the padded coded size; importing it as the display size would
        // rescale every sample.
        if info.coded_width < info.width || info.coded_height < info.height {
            return Err(BridgeError::InvalidInput(
                "decoded frame is larger than its allocated image",
            ));
        }
        let coded_width = u32::try_from(info.coded_width)
            .map_err(|_| BridgeError::InvalidInput("invalid coded width"))?;
        let coded_height = u32::try_from(info.coded_height)
            .map_err(|_| BridgeError::InvalidInput("invalid coded height"))?;
        let access = frame.lock_access()?;
        let mut status = 0;
        // SAFETY: the guard holds FFmpeg's frame and queue locks while C reads the frame.
        let step =
            unsafe { alpha2_native_gpu_vulkan_acquire_check(access.access(), &raw mut status) };
        if step != 0 {
            return Err(helper_error("import check", step, status));
        }
        let (device, allocator, queue_family) = {
            let access = access.access();
            (
                access.frame().device,
                access.allocator(),
                access.queue_family(),
            )
        };
        drop(access);
        let ticket = self.import_tickets.take(device, allocator, queue_family)?;
        let (texture_format, luma_format, chroma_format) = match layout {
            NativePlaneLayout::P010 => (
                wgpu::TextureFormat::P010,
                wgpu::TextureFormat::R16Unorm,
                wgpu::TextureFormat::Rg16Unorm,
            ),
            _ => (
                wgpu::TextureFormat::NV12,
                wgpu::TextureFormat::R8Unorm,
                wgpu::TextureFormat::Rg8Unorm,
            ),
        };
        let hal_descriptor = wgpu_hal::TextureDescriptor {
            label: Some("alpha2 FFmpeg decode surface"),
            size: wgpu::Extent3d {
                width: coded_width,
                height: coded_height,
                depth_or_array_layers: 1,
            },
            mip_level_count: 1,
            sample_count: 1,
            dimension: wgpu::TextureDimension::D2,
            format: texture_format,
            usage: wgpu::TextureUses::RESOURCE,
            memory_flags: wgpu_hal::MemoryFlags::empty(),
            view_formats: Vec::new(),
        };
        // SAFETY: this context is always Vulkan, and the identity check above established the
        // same logical device.
        let hal_device = unsafe { self.gpu.device().as_hal::<wgpu_hal::api::Vulkan>() }
            .ok_or(BridgeError::InvalidInput("renderer is not Vulkan"))?;
        let drop_owner = Arc::clone(&frame);
        // SAFETY: FFmpeg created the image with the validated descriptor and keeps it alive via
        // `drop_owner`; wgpu must not destroy it.
        let hal_texture = unsafe {
            hal_device.texture_from_raw(
                vk::Image::from_raw(info.images[0]),
                &hal_descriptor,
                Some(Box::new(move || drop(drop_owner))),
                wgpu_hal::vulkan::TextureMemory::External,
            )
        };
        let descriptor = wgpu::TextureDescriptor {
            label: Some("alpha2 FFmpeg decode surface"),
            size: hal_descriptor.size,
            mip_level_count: 1,
            sample_count: 1,
            dimension: wgpu::TextureDimension::D2,
            format: texture_format,
            usage: wgpu::TextureUsages::TEXTURE_BINDING,
            view_formats: &[],
        };
        // SAFETY: the batched acquire moves the whole image to the shader-read layout before
        // its first wgpu submission, and the HAL texture is this device's.
        let texture = unsafe {
            self.gpu
                .device()
                .create_texture_from_hal::<wgpu_hal::api::Vulkan>(
                    hal_texture,
                    &descriptor,
                    wgpu::TextureUses::RESOURCE,
                )
        };
        let y = texture.create_view(&wgpu::TextureViewDescriptor {
            label: Some("alpha2 FFmpeg decode Y"),
            format: Some(luma_format),
            aspect: wgpu::TextureAspect::Plane0,
            ..Default::default()
        });
        let uv = texture.create_view(&wgpu::TextureViewDescriptor {
            label: Some("alpha2 FFmpeg decode UV"),
            format: Some(chroma_format),
            aspect: wgpu::TextureAspect::Plane1,
            ..Default::default()
        });
        let pixels = u64::from(coded_width).saturating_mul(u64::from(coded_height));
        let lease = Arc::new(VulkanImportLease {
            frame,
            _texture: texture,
            gpu: self.gpu.clone(),
            batch: Arc::clone(&self.submission_batch),
            tickets: Arc::clone(&self.import_tickets),
            ticket: Some(ticket),
            staged_signals: Mutex::new(Vec::new()),
            queued: AtomicBool::new(false),
            acquired: AtomicBool::new(false),
            released: AtomicBool::new(false),
            retained_bytes: pixels
                .saturating_mul(layout.bytes_per_pixel_numerator())
                .div_ceil(2),
        });
        Ok(ImportedVulkanFrame {
            y,
            uv,
            layout,
            coded_size: [coded_width, coded_height],
            lease,
        })
    }
}

/// What one Vulkan adapter offers the shared device (queue families and video extensions), read
/// without opening it.
#[derive(Clone, Debug, PartialEq, Eq)]
struct VideoPlan {
    decode_family: u32,
    /// The video-encode queue family, when it has every extension `FFmpeg`'s encoder needs.
    encode_family: Option<u32>,
    video_extensions: Vec<&'static CStr>,
    decode_caps: vk::VideoCodecOperationFlagsKHR,
    /// The codecs the decode extensions cover (`h264`, `hevc`, `av1`, `vp9`).
    decode_codecs: Vec<&'static str>,
    /// The codecs the encode extensions cover; empty without an encode queue.
    encode_codecs: Vec<&'static str>,
}

/// Decode extensions the shared device enables where available: extension, the operation
/// `FFmpeg` is told the decode queue does, and the codec name.
const DECODE_EXTENSIONS: [(&CStr, vk::VideoCodecOperationFlagsKHR, &str); 4] = [
    (
        ash::khr::video_decode_h264::NAME,
        vk::VideoCodecOperationFlagsKHR::DECODE_H264,
        "h264",
    ),
    (
        ash::khr::video_decode_h265::NAME,
        vk::VideoCodecOperationFlagsKHR::DECODE_H265,
        "hevc",
    ),
    (
        ash::khr::video_decode_av1::NAME,
        vk::VideoCodecOperationFlagsKHR::DECODE_AV1,
        "av1",
    ),
    (VIDEO_DECODE_VP9_NAME, DECODE_VP9, "vp9"),
];

#[derive(Clone, Copy, Debug)]
struct QueueFamily {
    flags: vk::QueueFlags,
    count: u32,
}

/// Whether the shared device can be made on an adapter with `families`, the extensions
/// `supports` answers for, and NV12 plane views (`nv12`), and with what queues and extensions.
///
/// The graphics queue must be family 0 (wgpu-hal's shared-device path) and carry the compute
/// and transfer work `FFmpeg` submits to it. Decode needs its own queue family and at least
/// H.264. Encode is optional.
fn video_plan(
    families: &[QueueFamily],
    supports: impl Fn(&CStr) -> bool,
    nv12: bool,
) -> Result<VideoPlan, String> {
    if !nv12 {
        return Err("Vulkan adapter does not support NV12 plane views".to_owned());
    }
    let family = |flag: vk::QueueFlags| {
        families
            .iter()
            .position(|family| family.flags.contains(flag) && family.count > 0)
            .and_then(|index| u32::try_from(index).ok())
    };
    let graphics_family =
        family(vk::QueueFlags::GRAPHICS).ok_or("Vulkan adapter has no graphics queue")?;
    if graphics_family != 0 {
        return Err("wgpu-hal shared-device path requires graphics queue family 0".to_owned());
    }
    if !families[0]
        .flags
        .contains(vk::QueueFlags::COMPUTE | vk::QueueFlags::TRANSFER)
    {
        return Err("graphics queue family lacks FFmpeg compute/transfer capabilities".to_owned());
    }
    let decode_family = family(vk::QueueFlags::VIDEO_DECODE_KHR)
        .ok_or("Vulkan adapter has no video-decode queue")?;
    if decode_family == graphics_family {
        return Err(
            "shared Vulkan device requires a decode queue separate from wgpu graphics".to_owned(),
        );
    }
    let base = [
        ash::khr::video_queue::NAME,
        ash::khr::video_decode_queue::NAME,
    ];
    if !base.iter().all(|extension| supports(extension))
        || !supports(ash::khr::video_decode_h264::NAME)
    {
        return Err("Vulkan H.264 video decode is unavailable".to_owned());
    }
    let mut video_extensions = base.to_vec();
    let mut decode_caps = vk::VideoCodecOperationFlagsKHR::empty();
    let mut decode_codecs = Vec::new();
    for (extension, operation, codec) in DECODE_EXTENSIONS {
        if supports(extension) {
            video_extensions.push(extension);
            decode_caps |= operation;
            decode_codecs.push(codec);
        }
    }
    // FFmpeg's Vulkan encoders need the encode queue and `video_maintenance1` (sessions without
    // pre-allocated DPB images), neither of which wgpu enables.
    let encode_base = [
        ash::khr::video_encode_queue::NAME,
        ash::khr::video_maintenance1::NAME,
    ];
    let encoded: Vec<(&'static str, &'static CStr)> = ENCODE_EXTENSIONS
        .iter()
        .filter(|(_, extension, _)| supports(extension))
        .map(|(codec, extension, _)| (*codec, *extension))
        .collect();
    let encode_family = family(vk::QueueFlags::VIDEO_ENCODE_KHR)
        .filter(|_| encode_base.iter().all(|extension| supports(extension)) && !encoded.is_empty());
    let encode_codecs = if encode_family.is_some() {
        video_extensions.extend_from_slice(&encode_base);
        video_extensions.extend(encoded.iter().map(|(_, extension)| *extension));
        encoded.iter().map(|(codec, _)| *codec).collect()
    } else {
        Vec::new()
    };
    Ok(VideoPlan {
        decode_family,
        encode_family,
        video_extensions,
        decode_caps,
        decode_codecs,
        encode_codecs,
    })
}

/// [`video_plan`] for `exposed`, read from its instance.
fn adapter_video_plan(
    instance: &ash::Instance,
    exposed: &wgpu_hal::ExposedAdapter<wgpu_hal::api::Vulkan>,
) -> Result<VideoPlan, String> {
    let physical = exposed.adapter.raw_physical_device();
    // SAFETY: `physical` was enumerated from this instance.
    let families: Vec<QueueFamily> =
        unsafe { instance.get_physical_device_queue_family_properties(physical) }
            .iter()
            .map(|family| QueueFamily {
                flags: family.queue_flags,
                count: family.queue_count,
            })
            .collect();
    // SAFETY: `physical` was enumerated from this instance.
    let extensions = unsafe { instance.enumerate_device_extension_properties(physical) }
        .map_err(|error| format!("Vulkan extensions: {error}"))?;
    let supports = |name: &CStr| {
        extensions.iter().any(|property| {
            property
                .extension_name_as_c_str()
                .is_ok_and(|candidate| candidate == name)
        })
    };
    video_plan(
        &families,
        supports,
        exposed
            .features
            .contains(wgpu::Features::TEXTURE_FORMAT_NV12),
    )
}

/// One Vulkan adapter as the shared device's choice sees it, read without opening a device.
#[derive(Clone, Debug)]
pub struct VulkanCandidate {
    /// Position in the Vulkan enumeration, as `ALPHA2_GPU_ADAPTER` and
    /// [`SharedVulkanContext::create_on`] use it.
    pub index: usize,
    /// Position in the loader's physical-device order, which `FFmpeg` takes for its own device.
    pub loader_index: Option<usize>,
    /// The adapter's name, which a name pin matches.
    pub name: String,
    /// Its kind, for the high-performance order.
    pub device_type: wgpu::DeviceType,
    /// Whether the shared device can be made on it, or why not.
    pub shares: Result<(), String>,
    /// The codecs its decode extensions cover.
    pub decode_codecs: Vec<&'static str>,
    /// The codecs its encode queue and extensions cover.
    pub encode_codecs: Vec<&'static str>,
}

/// Every Vulkan adapter, in enumeration order, with what it offers the shared device.
///
/// # Errors
/// Returns why there is no Vulkan instance.
pub fn vulkan_candidates() -> Result<Vec<VulkanCandidate>, SharedVulkanError> {
    let _turn = alpha2_gpu::serialize_acquisition();
    let hal_instance = hal_instance()?;
    // SAFETY: adapter enumeration does not require a surface.
    let adapters = unsafe { hal_instance.enumerate_adapters(None) };
    Ok(adapters
        .iter()
        .enumerate()
        .map(|(index, exposed)| {
            let instance = exposed.adapter.shared_instance().raw_instance();
            let physical = exposed.adapter.raw_physical_device();
            // SAFETY: the instance is alive for the call; enumeration has no other precondition.
            let loader_index = unsafe { instance.enumerate_physical_devices() }
                .ok()
                .and_then(|order| order.iter().position(|device| *device == physical));
            let plan = adapter_video_plan(instance, exposed);
            VulkanCandidate {
                index,
                loader_index,
                name: exposed.info.name.clone(),
                device_type: exposed.info.device_type,
                decode_codecs: plan
                    .as_ref()
                    .map(|plan| plan.decode_codecs.clone())
                    .unwrap_or_default(),
                encode_codecs: plan
                    .as_ref()
                    .map(|plan| plan.encode_codecs.clone())
                    .unwrap_or_default(),
                shares: plan.map(|_| ()),
            }
        })
        .collect())
}

/// The record the adapter choice ranks `candidate` by. GPU decode and encode both need the
/// shared device, so an adapter that cannot carry it does neither.
fn candidate_record(
    candidate: &VulkanCandidate,
    decode_allowed: bool,
) -> alpha2_gpu::AdapterRecord {
    let mut record = alpha2_gpu::AdapterRecord::unasked(
        candidate.index,
        candidate.name.clone(),
        "vulkan",
        candidate.device_type == wgpu::DeviceType::Cpu,
    );
    record.renders = Ok(());
    record.decodes = match (&candidate.shares, decode_allowed) {
        (Err(reason), _) => Err(reason.clone()),
        (Ok(()), false) => Err("decode is in software".to_owned()),
        (Ok(()), true) => Ok(()),
    };
    record.gpu_encodes = alpha2_gpu::GPU_ENCODE_CODECS
        .iter()
        .map(|&codec| {
            let answer = match &candidate.shares {
                Err(reason) => Err(reason.clone()),
                Ok(()) if candidate.encode_codecs.contains(&codec) => Ok(()),
                Ok(()) => Err(format!(
                    "{} has no Vulkan video encoder for {codec}",
                    candidate.name
                )),
            };
            (codec, answer)
        })
        .collect();
    record
}

/// The shared device the adapter choice made, its report, and why there is none if so.
pub struct SharedVulkanChoice<T = SharedVulkanContext> {
    /// The shared device, when one was made.
    pub context: Option<T>,
    /// Which adapter the policy took and why, with every Vulkan adapter's record.
    pub report: alpha2_gpu::AdapterChoiceReport,
    /// Why no shared device was made although one was wanted.
    pub unavailable: Option<String>,
}

/// Makes the shared Vulkan device on the adapter `policy` takes, ranked as every host ranks
/// adapters ([`alpha2_gpu::rank_adapters`]): under `auto`, the adapter encoding the most codecs
/// on the GPU, then one that decodes when `decode_allowed`, then high-performance order.
/// Without `wanted` nothing is made and the report says where the product device renders.
///
/// If making the device fails, the failure goes into that adapter's record and the choice runs
/// again: `auto` tries the next capable adapter, while `performance` and a pin keep their
/// adapter and render without a shared device.
///
/// # Errors
/// Returns why there is no Vulkan instance to choose on.
pub fn choose_shared_vulkan(
    policy: &alpha2_gpu::AdapterPolicy,
    decode_allowed: bool,
    wanted: bool,
) -> Result<SharedVulkanChoice, SharedVulkanError> {
    let candidates = vulkan_candidates()?;
    Ok(choose_over(
        candidates,
        policy,
        decode_allowed,
        wanted,
        |index| SharedVulkanContext::create_on(index).map_err(|error| error.to_string()),
    ))
}

/// [`choose_shared_vulkan`] over `candidates`, making the device with `create_on`.
fn choose_over<T>(
    mut candidates: Vec<VulkanCandidate>,
    policy: &alpha2_gpu::AdapterPolicy,
    decode_allowed: bool,
    wanted: bool,
    mut create_on: impl FnMut(usize) -> Result<T, String>,
) -> SharedVulkanChoice<T> {
    // In product request order, so ties render where the product device would.
    let kinds: Vec<(usize, wgpu::DeviceType)> = candidates
        .iter()
        .enumerate()
        .map(|(at, candidate)| (at, candidate.device_type))
        .collect();
    let order = alpha2_gpu::product_order(&kinds);
    let mut ordered: Vec<Option<VulkanCandidate>> = candidates.drain(..).map(Some).collect();
    let mut candidates: Vec<VulkanCandidate> = order
        .into_iter()
        .filter_map(|at| ordered[at].take())
        .collect();
    loop {
        let records: Vec<alpha2_gpu::AdapterRecord> = candidates
            .iter()
            .map(|candidate| candidate_record(candidate, decode_allowed))
            .collect();
        let ranked = match alpha2_gpu::rank_adapters(&records, policy, decode_allowed) {
            Ok(ranked) => ranked,
            Err(reason) => {
                return SharedVulkanChoice {
                    context: None,
                    report: alpha2_gpu::AdapterChoiceReport {
                        policy: policy.clone(),
                        chosen: None,
                        reason: alpha2_gpu::ChoiceReason::NoSharedDevice,
                        explanation: reason.clone(),
                        records,
                    },
                    unavailable: wanted.then_some(reason),
                };
            }
        };
        let mut report = alpha2_gpu::AdapterChoiceReport::ranked(policy.clone(), records, &ranked);
        let first = ranked.order[0];
        if !wanted {
            return SharedVulkanChoice {
                context: None,
                report,
                unavailable: None,
            };
        }
        let candidate = &mut candidates[first];
        if let Err(reason) = &candidate.shares {
            let unavailable = format!(
                "{} cannot share a device with the video decoder and encoder: {reason}",
                candidate.name
            );
            if !policy.is_pin() {
                report.reason = alpha2_gpu::ChoiceReason::NoSharedDevice;
            }
            report.explanation = format!("{}; {unavailable}", report.explanation);
            return SharedVulkanChoice {
                context: None,
                report,
                unavailable: Some(unavailable),
            };
        }
        match create_on(candidate.index) {
            Ok(context) => {
                return SharedVulkanChoice {
                    context: Some(context),
                    report,
                    unavailable: None,
                };
            }
            Err(reason) => candidate.shares = Err(format!("the shared device failed: {reason}")),
        }
    }
}

/// Whether `ALPHA2_GPU_ADAPTER` pins one adapter, which probes opening their own devices then
/// keep to.
#[must_use]
pub fn adapter_pinned() -> bool {
    alpha2_gpu::AdapterPolicy::from_env().is_pin()
}

/// The loader physical-device index of the adapter the host's choice makes the shared device
/// on: the `device` index `FFmpeg`'s `av_hwdevice_ctx_create` takes. Probes that open their own
/// device pass it, so they use the device an export runs on.
///
/// # Errors
/// Returns why no device is chosen: no Vulkan instance, or no adapter the policy takes that
/// could carry the shared device.
pub fn shared_device_index() -> Result<usize, SharedVulkanError> {
    let candidates = vulkan_candidates()?;
    let mut chosen = None;
    let choice = choose_over(
        candidates.clone(),
        &alpha2_gpu::AdapterPolicy::from_env(),
        true,
        true,
        |index| {
            chosen = Some(index);
            Ok(())
        },
    );
    let index = chosen.ok_or_else(|| {
        SharedVulkanError(
            choice
                .unavailable
                .unwrap_or_else(|| "no Vulkan adapter could carry the shared device".to_owned()),
        )
    })?;
    candidates
        .iter()
        .find(|candidate| candidate.index == index)
        .and_then(|candidate| candidate.loader_index)
        .ok_or_else(|| SharedVulkanError("the chosen adapter is not in the device list".into()))
}

#[cfg(test)]
#[path = "vulkan_failure_tests.rs"]
mod failure_tests;

#[cfg(test)]
#[path = "vulkan_tests.rs"]
mod tests;