Isolated owner process for producer-backed video interop leases.
One owner should be started per producing element or native buffer pool. Its mailbox is reserved for lease lifecycle messages so media traffic in the producer cannot delay buffer retirement.
Issuance has an explicit ownership boundary. Capacity is reserved before the
backend token is sent. The caller owns the token until the reservation is
accepted and the token-bearing commit message is sent. Every later error is
tagged :transferred, and the owner (or the token's independent owner-crash
destructor) is responsible for cleanup.
Release callbacks run on one monitored serial executor so arbitrary callback latency cannot block the owner mailbox. Callbacks used with automatic retry must be idempotent for one backend token. Retry is single-flight per public lease token. Failed entries remain alive and retryable after producer death or retry exhaustion; there is no implicit fatal policy based on observer liveness.
A producer-supplied abandonment_guard_factory may attach one unique,
authority-validated native resource envelope to every root and retained holder.
Guard construction is transactional: no holder is published when construction fails. Guard
destructor messages are fallback releases with separate abandonment
accounting and are idempotent with the deterministic explicit release path.
Ordered release tombstones are bounded; releases older than retained history
are reported as unclassified instead of being guessed to be duplicates.
Summary
Functions
Stops accepting new issues and retains, then drains existing holders.
Stops admission and waits for all holders and release callbacks.
Transfers a private backend token to the owner and returns a confirmed lease.
Retries a failed final backend release without exiting the caller.
Starts a lease owner linked to the producing process.
Starts a supervisor-owned lease owner that monitors a distinct producer.
Returns lease counts, release timings, and mailbox depth.
Types
@type option() :: {:producer, pid()} | {:release, release_callback()} | {:release_retry, retry_policy()} | {:max_active, pos_integer() | :infinity} | {:notify, pid() | nil} | {:notify_releases, boolean()} | {:abandonment_guard_factory, abandonment_guard_factory() | nil}
@type ownership_error() :: {:caller_owned | :transferred, term()}
@type retry_policy() :: :manual | {:exponential, initial_ms: pos_integer(), max_ms: pos_integer(), max_attempts: pos_integer() | :infinity}
@type stats() :: %{ state: :open | :draining, active_leases: non_neg_integer(), issue_reservations: non_neg_integer(), active_holders: non_neg_integer(), oldest_lease_age_ns: non_neg_integer() | nil, issued_leases: non_neg_integer(), issued_holders: non_neg_integer(), retain_requests: non_neg_integer(), retain_cancellations: non_neg_integer(), explicit_releases: non_neg_integer(), abandonments: non_neg_integer(), late_releases_after_abandonment: non_neg_integer(), duplicate_releases: non_neg_integer(), unclassified_releases: non_neg_integer(), release_tombstone_evictions: non_neg_integer(), abandonment_tombstone_evictions: non_neg_integer(), release_callbacks: non_neg_integer(), release_failures: non_neg_integer(), release_retries: non_neg_integer(), release_callback_total_ns: non_neg_integer(), release_callback_max_ns: non_neg_integer(), release_executor_queue_depth: non_neg_integer(), release_executor_active_age_ns: non_neg_integer() | nil, release_executor_restarts: non_neg_integer(), malformed_messages: non_neg_integer(), release_tombstones: non_neg_integer(), release_tombstone_limit: pos_integer(), drain_waiters: non_neg_integer(), message_queue_len: non_neg_integer() }
Functions
@spec close(pid(), timeout()) :: :ok | {:ok, :draining} | {:error, :timeout | {:owner_down, term()}}
Stops accepting new issues and retains, then drains existing holders.
@spec drain(pid(), timeout()) :: :ok | {:error, :timeout | {:owner_down, term()} | {:release_failed, reference(), term()}}
Stops admission and waits for all holders and release callbacks.
A timeout removes only this waiter and leaves the owner draining. A failed
final callback returns its public token so retry/3 can address it.
@spec issue(pid(), term(), keyword()) :: {:ok, VideoInterop.Lease.t()} | {:error, ownership_error()}
Transfers a private backend token to the owner and returns a confirmed lease.
The owner is monitored and capacity is reserved before the backend token is
sent. Capacity, draining, timeout, or owner death before the token-bearing
commit are :caller_owned errors. Sending that commit is the transfer
boundary; every later timeout, release failure, or owner death is a
:transferred error.
Because a local PID can die concurrently with send, backend tokens must also have an owner-crash/message-drop destructor fallback.
Retries a failed final backend release without exiting the caller.
@spec start_link([option()]) :: GenServer.on_start()
Starts a lease owner linked to the producing process.
This intentionally uses an ordinary link rather than an OTP parent link so the
owner can trap producer exit and outlive it until every lease drains.
release_retry defaults to :manual. Automatic exponential retry requires an
idempotent release callback. Before stopping after successful drainage, the
owner sends its final stats followed by the two-field drained notification.
@spec start_supervised([option()]) :: GenServer.on_start()
Starts a supervisor-owned lease owner that monitors a distinct producer.
Returns lease counts, release timings, and mailbox depth.