VideoInterop.LeaseOwner (video_interop v0.1.0)

Copy Markdown View Source

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

abandonment_guard_factory()

@type abandonment_guard_factory() ::
  (pid(), reference(), reference() ->
     {:ok, VideoInterop.AbandonmentGuard.t()} | {:error, term()})
  | {module(), atom(), [term()]}

option()

@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}

ownership_error()

@type ownership_error() :: {:caller_owned | :transferred, term()}

release_callback()

@type release_callback() :: (term() -> term()) | {module(), atom(), [term()]}

retry_policy()

@type retry_policy() ::
  :manual
  | {:exponential,
     initial_ms: pos_integer(),
     max_ms: pos_integer(),
     max_attempts: pos_integer() | :infinity}

stats()

@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

close(owner, timeout \\ 5000)

@spec close(pid(), timeout()) ::
  :ok | {:ok, :draining} | {:error, :timeout | {:owner_down, term()}}

Stops accepting new issues and retains, then drains existing holders.

drain(owner, timeout \\ 5000)

@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.

issue(owner, backend_token, opts \\ [])

@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.

retry(owner, token, timeout \\ 5000)

@spec retry(pid(), reference(), timeout()) :: :ok | {:error, term()}

Retries a failed final backend release without exiting the caller.

start_link(opts)

@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.

start_supervised(opts)

@spec start_supervised([option()]) :: GenServer.on_start()

Starts a supervisor-owned lease owner that monitors a distinct producer.

stats(owner, timeout \\ 5000)

@spec stats(pid(), timeout()) :: stats() | {:error, :timeout | {:owner_down, term()}}

Returns lease counts, release timings, and mailbox depth.