backpressure.c

Go to the documentation of this file.

#include <ascii-chat/network/acip/security.h>
#include <ascii-chat/otel.h>
#include <ascii-chat/network/acip/backpressure.h>
#include <ascii-chat/network/acip/transport.h>
#include <ascii-chat/network/packet/packet.h>
#include <libplatform/cond.h>
#include <libplatform/thread.h>
#include <ascii-chat/buffer.h>
#include <ascii-chat/common.h>
#include <ascii-chat/util/endian.h>
#include <ascii-chat/util/time.h>
#include <string.h>
 
#define OUTBOUND_MAX_PACKETS 512
#define OUTBOUND_MAX_STREAMS 64
#define WIRE_OVERHEAD (sizeof(packet_header_t) + 32 + CRYPTO_NONCE_SIZE + CRYPTO_MAC_SIZE)

typedef struct outbound_packet {
  struct outbound_packet *next;
  uint8_t *data;
  size_t size, cost;
  uint64_t created;
  uint16_t type;
  uint32_t stream;
  acip_send_options_t options;
  bool video, encoded, keyframe, audio, wait, done;
  asciichat_error_t result;
  gotel_context_t telemetry_context;
} outbound_packet_t;
 
typedef struct {
  uint16_t type;
  uint32_t stream;
  uint64_t last_used_ns;
  atomic_t needs_keyframe;
} outbound_stream_t;
 
struct acip_outbound {
  acip_transport_t *transport;
  mutex_t mutex;
  cond_t changed;
  platform_thread_t worker;
  bool stopping, failed, stopped_by_caller;
  asciichat_error_t failure_code;
  asciichat_error_context_t failure;
  asciichat_error_context_t failure_detail;
  outbound_packet_t *head;
  outbound_packet_t *recycled;
  size_t recycled_count;
  size_t count, bytes, active_bytes;
  outbound_stream_t streams[OUTBOUND_MAX_STREAMS];
  atomic_t stream_count;
  atomic_t accepted, submitted, dropped, congestion, recovery_frames;
  atomic_t peak_bytes;
};
 
static size_t backend_bytes(acip_outbound_t *out) {
  return out->transport->methods->queued_bytes ? out->transport->methods->queued_bytes(out->transport) : 0;
}
 
static outbound_stream_t *stream_state(acip_outbound_t *out, uint16_t type, uint32_t id) {
  size_t stream_count = (size_t)atomic_load_u64(&out->stream_count);
  for (size_t i = 0; i < stream_count; ++i)
    if (out->streams[i].type == type && out->streams[i].stream == id) {
      out->streams[i].last_used_ns = time_get_ns();
      return &out->streams[i];
    }
  if (stream_count == OUTBOUND_MAX_STREAMS) {
    size_t victim = OUTBOUND_MAX_STREAMS;
    for (size_t i = 0; i < stream_count; ++i) {
      bool pending = false;
      for (outbound_packet_t *p = out->head; p; p = p->next)
        if (p->encoded && p->type == out->streams[i].type && p->stream == out->streams[i].stream) {
          pending = true;
          break;
        }
      if (!pending &&
          (victim == OUTBOUND_MAX_STREAMS || out->streams[i].last_used_ns < out->streams[victim].last_used_ns))
        victim = i;
    }
    if (victim == OUTBOUND_MAX_STREAMS) {
      log_warn("Outbound stream state limit reached (%u); all stream slots are active", OUTBOUND_MAX_STREAMS);
      return NULL;
    }
    log_warn("Evicting inactive outbound stream state for type %u stream %u", out->streams[victim].type,
             out->streams[victim].stream);
    outbound_stream_t *stream = &out->streams[victim];
    *stream = (outbound_stream_t){.type = type, .stream = id, .last_used_ns = time_get_ns()};
    atomic_store_bool(&stream->needs_keyframe, true);
    return stream;
  }
  size_t index = stream_count;
  outbound_stream_t *stream = &out->streams[index];
  *stream = (outbound_stream_t){.type = type, .stream = id, .last_used_ns = time_get_ns()};
  atomic_store_bool(&stream->needs_keyframe, true);
  // Publish immutable identity only after initialization. The keyframe bit is
  // atomic because it is updated independently of the stream identity.
  atomic_store_u64(&out->stream_count, index + 1);
  return stream;
}
 
// Keep a small connection-owned working set instead of visiting the global
// allocator and buffer domain for every media packet. All access holds mutex.
static void packet_free(acip_outbound_t *out, outbound_packet_t *packet) {
  if (!out->stopping && !out->failed && out->recycled_count < 8 && buffer_capacity(packet->data) <= 256 * 1024) {
    packet->next = out->recycled;
    out->recycled = packet;
    ++out->recycled_count;
    return;
  }
  buffer_release(packet->data);
  SAFE_FREE(packet);
}
 
static bool same_stream(const outbound_packet_t *a, const outbound_packet_t *b) {
  return a->type == b->type && a->stream == b->stream;
}
 
static bool outbound_insert_packet(acip_outbound_t *out, outbound_packet_t *packet, bool priority) {
  outbound_packet_t **link = &out->head;
  if (priority) {
    for (outbound_packet_t **scan = &out->head; *scan; scan = &(*scan)->next)
      if (!(*scan)->video && !(*scan)->audio && (*scan)->type != PACKET_TYPE_PING && (*scan)->type != PACKET_TYPE_PONG)
        link = &(*scan)->next;
  }
  while (*link && (!priority || !(*link)->video))
    link = &(*link)->next;
  bool was_empty = out->head == NULL;
  packet->next = *link;
  *link = packet;
  return was_empty;
}
 
/* Conservatively invalidate all unsent encoded output for this stream. The
 * producer requests an IDR with parameter sets before resuming its chain. */
static void discard_stream(acip_outbound_t *out, const outbound_packet_t *video) {
  if (video->encoded) {
    outbound_stream_t *stream = stream_state(out, video->type, video->stream);
    if (stream)
      atomic_store_bool(&stream->needs_keyframe, true);
  }
  outbound_packet_t **link = &out->head;
  while (*link) {
    outbound_packet_t *packet = *link;
    if (packet->video && same_stream(packet, video)) {
      *link = packet->next;
      out->bytes -= packet->cost;
      --out->count;
      atomic_fetch_add_u64(&out->dropped, 1);
      ascii_otel_metric(ASCII_OTEL_METRIC_OUTBOUND_DROP, 1);
      packet_free(out, packet);
    } else {
      link = &packet->next;
    }
  }
}
 
static void expire_video(acip_outbound_t *out, uint64_t now) {
  for (;;) {
    outbound_packet_t *expired = NULL;
    for (outbound_packet_t *p = out->head; p; p = p->next)
      if (p->video && now - p->created > ACIP_OUTBOUND_VIDEO_AGE_NS) {
        expired = p;
        break;
      }
    if (!expired)
      break;
    outbound_packet_t identity = *expired;
    discard_stream(out, &identity);
  }
}
 
static void expire_datagrams(acip_outbound_t *out, uint64_t now) {
  outbound_packet_t **link = &out->head;
  while (*link) {
    outbound_packet_t *packet = *link;
    if (packet->options.delivery != ACIP_DELIVERY_DATAGRAM || !packet->options.deadline_ns ||
        packet->options.deadline_ns > now) {
      link = &packet->next;
      continue;
    }
    if (packet->encoded) {
      outbound_packet_t identity = *packet;
      discard_stream(out, &identity);
      link = &out->head;
      continue;
    }
    *link = packet->next;
    out->bytes -= packet->cost;
    --out->count;
    atomic_fetch_add_u64(&out->dropped, 1);
    ascii_otel_metric(ASCII_OTEL_METRIC_OUTBOUND_DROP, 1);
    packet_free(out, packet);
  }
}
 
// Keep the terminal wrapper plus its relevant nested cause. Follow causal IDs,
// not merely older registry entries, and never cross this send's checkpoint.
static asciichat_error_context_t backend_failure_detail(asciichat_errno_scope_t scope,
                                                        const asciichat_error_context_t *cause) {
  asciichat_error_context_t frames[ASCIICHAT_ERRNO_MAX_DEPTH];
  size_t count = errno_snapshot(scope.thread_id, frames, ASCIICHAT_ERRNO_MAX_DEPTH).copied;
  asciichat_error_context_t detail = {0};
  uint64_t next = cause->cause_id;
  for (size_t i = 0; i < count && next; ++i) {
    if (frames[i].generation != scope.generation || frames[i].error_id <= scope.after_id)
      continue;
    if (frames[i].error_id != next)
      continue;
    if (!detail.has_system_error && !detail.has_wsa_error)
      detail = frames[i];
    else if (frames[i].has_system_error || frames[i].has_wsa_error)
      detail = frames[i];
    next = frames[i].cause_id;
  }
  return detail;
}
 
static asciichat_error_t import_failure(const asciichat_error_context_t *cause,
                                        const asciichat_error_context_t *detail) {
  if (detail->code != ASCIICHAT_OK && !cause->has_system_error && !cause->has_wsa_error) {
    errno_import(detail);
    // Recreate the wrapper at its original source location. Normal recording
    // links it to the newly imported cause in this producer's registry; simply
    // importing both would leave the wrapper pointing at the handled worker ID.
    return errno_set_with_message(cause->code, cause->file, cause->line, cause->function, "%s", cause->context_message);
  }
  return errno_import(cause);
}
 
// The caller holds mutex. Retain the first terminal cause, reject later
// admissions, and wake queued producers with that failure. Backend close is
// performed by the worker after releasing mutex.
static void outbound_record_failure_and_fail_pending(acip_outbound_t *out, asciichat_error_t code,
                                                     const asciichat_error_context_t *cause,
                                                     const asciichat_error_context_t *detail) {
  if (!out->failed) {
    out->failure_code = code;
    if (cause)
      out->failure = *cause;
    if (detail)
      out->failure_detail = *detail;
  }
  out->failed = true;
  while (out->head) {
    outbound_packet_t *p = out->head;
    out->head = p->next;
    out->bytes -= p->cost;
    --out->count;
    if (p->wait) {
      p->result = out->failure_code;
      p->done = true;
    } else {
      packet_free(out, p);
    }
  }
  cond_broadcast(&out->changed);
}
 
static asciichat_error_t outbound_send(acip_outbound_t *out, const outbound_packet_t *packet) {
  ASCII_OTEL_SCOPE_PARENT(ACIP_OUTBOUND_SEND, &packet->telemetry_context);
  ASCII_OTEL_PACKET(packet->type, packet->size >= sizeof(packet_header_t) ? packet->size - sizeof(packet_header_t) : 0,
                    "send", acip_transport_get_type(out->transport));
  uint64_t wait_ns = time_get_ns() - packet->created;
  ASCII_OTEL_ATTR_INT("network.queue.wait_ns", wait_ns);
  ASCII_OTEL_ATTR_STRING("network.packet.category", packet->video ? "video" : packet->audio ? "audio" : "control");
  ascii_otel_metric(ASCII_OTEL_METRIC_OUTBOUND_WAIT, (double)wait_ns / 1e9);
  if (out->transport->methods->sendv_ex) {
    socket_buffer_t buffer = {packet->data, packet->size};
    return out->transport->methods->sendv_ex(out->transport, &buffer, 1, &packet->options);
  }
  return out->transport->methods->send(out->transport, packet->data, packet->size);
}
 
static void *outbound_worker(void *arg) {
  acip_outbound_t *out = arg;
  mutex_lock(&out->mutex);
  while (!out->stopping && !out->failed) {
    uint64_t now = time_get_ns();
    expire_video(out, now);
    expire_datagrams(out, now);
    outbound_packet_t *p = out->head;
    if (!p) {
      cond_timedwait(&out->changed, &out->mutex, 100 * NS_PER_MS_INT);
      continue;
    }
    // Rotation uses this same serialized writer for every transport. Flush
    // only when work is queued; idle polling must not reacquire the queue lock.
    mutex_unlock(&out->mutex);
    asciichat_error_t security_result = acip_security_flush(out->transport);
    mutex_lock(&out->mutex);
    if (security_result != ASCIICHAT_OK) {
      outbound_record_failure_and_fail_pending(out, security_result, NULL, NULL);
      break;
    }
    now = time_get_ns();
    p = out->head;
    if (!p)
      continue;
    /* Admitted reliable traffic expires by closing the connection, never by
     * silently removing an audio or control packet. */
    uint64_t deadline = p->audio ? 5 * NS_PER_SEC_INT : 10 * NS_PER_SEC_INT;
    /* Reliable expiry is terminal for control traffic only. Video and audio
     * are media queues and are governed by their media admission/expiry
     * policy; they must not tear down the peer merely because a slow backend
     * delayed submission. */
    if (!p->video && !p->audio && p->options.delivery == ACIP_DELIVERY_RELIABLE && now - p->created > deadline) {
      asciichat_errno_scope_t scope = errno_scope_begin();
      asciichat_error_context_t cause;
      SET_ERRNO(ERROR_NETWORK_TIMEOUT, "Outbound reliable packet expired before submission");
      bool has_cause = errno_peek_since(scope, &cause);
      outbound_record_failure_and_fail_pending(out, ERROR_NETWORK_TIMEOUT, has_cause ? &cause : NULL, NULL);
      errno_scope_end(scope, ASCIICHAT_ERRNO_HANDLED);
      break;
    }
    // Include backend-owned bytes before reserving room for audio and control.
    size_t backend = backend_bytes(out);
    // Keep scheduling decisions in the plaintext queue. Even tiny reliable
    // packets must not accumulate behind video in a backend's opaque FIFO.
    acip_transport_capabilities_t capabilities = {0};
    asciichat_error_t capability_result = acip_transport_get_capabilities(out->transport, &capabilities);
    if (capability_result != ASCIICHAT_OK) {
      outbound_record_failure_and_fail_pending(out, capability_result, NULL, NULL);
      break;
    }
    /* Media must wait for opaque backends without independent streams so it
     * cannot build an unbounded FIFO behind an already-buffered frame. A
     * reliable control packet is different: withholding it here can make it
     * expire while waiting behind the media backlog it is meant to control.
     * The backend preserves ordering, so submit control traffic and let it
     * drain normally. */
    if (backend != 0 && !capabilities.independent_streams && (p->video || p->audio)) {
      if (out->transport->methods->wait_writable) {
        mutex_unlock(&out->mutex);
        asciichat_error_t writable = out->transport->methods->wait_writable(out->transport, 10 * NS_PER_MS_INT);
        mutex_lock(&out->mutex);
        if (writable != ASCIICHAT_OK && writable != ERROR_NETWORK_TIMEOUT) {
          outbound_record_failure_and_fail_pending(out, writable, NULL, NULL);
          break;
        }
      } else {
        cond_timedwait(&out->changed, &out->mutex, 2 * NS_PER_MS_INT);
      }
      continue;
    }
    out->head = p->next;
    out->bytes -= p->cost;
    // Admission reserves plaintext plus two wire owners. During handoff one
    // wire owner moves into the backend's counter, so do not count it twice.
    out->active_bytes = p->cost - (p->size + WIRE_OVERHEAD);
    mutex_unlock(&out->mutex);
    /* One worker fixes wire/encryption order. Backends receive only complete
     * packets and must finish a partially submitted packet or fail the stream. */
    asciichat_errno_scope_t scope = errno_scope_begin();
    asciichat_error_t result = outbound_send(out, p);
    asciichat_error_context_t cause = {0};
    asciichat_error_context_t detail = {0};
    if (result != ASCIICHAT_OK) {
      // Only this send's matching failure belongs to the submission. A backend
      // may return a status without context, or leave an unrelated nested error.
      if (!errno_peek_since(scope, &cause) || cause.code != result) {
        SET_ERRNO(result, "Outbound backend submission failed");
        errno_peek_since(scope, &cause);
      } else {
        detail = backend_failure_detail(scope, &cause);
      }
    }
    errno_scope_end(scope, ASCIICHAT_ERRNO_HANDLED);
    bool datagram_drop =
        p->options.delivery == ACIP_DELIVERY_DATAGRAM &&
        (result == ERROR_NETWORK_TIMEOUT || result == ERROR_BUFFER_FULL || result == ERROR_NOT_SUPPORTED);
    if (!p->wait) {
      mutex_lock(&out->mutex);
      out->active_bytes = 0;
      --out->count;
      if (datagram_drop) {
        if (p->encoded)
          discard_stream(out, p);
        atomic_fetch_add_u64(&out->dropped, 1);
        ascii_otel_metric(ASCII_OTEL_METRIC_OUTBOUND_DROP, 1);
      }
      if (result == ASCIICHAT_OK) {
        atomic_fetch_add_u64(&out->submitted, 1);
        if (p->encoded && p->keyframe)
          atomic_fetch_add_u64(&out->recovery_frames, 1);
      }
      if (result != ASCIICHAT_OK && !datagram_drop)
        outbound_record_failure_and_fail_pending(out, result, &cause, &detail);
      cond_broadcast(&out->changed);
      packet_free(out, p);
      mutex_unlock(&out->mutex);
      if (result != ASCIICHAT_OK && !datagram_drop)
        break;
      continue;
    }
    mutex_lock(&out->mutex);
    out->active_bytes = 0;
    --out->count;
    if (datagram_drop) {
      if (p->encoded)
        discard_stream(out, p);
      atomic_fetch_add_u64(&out->dropped, 1);
      ascii_otel_metric(ASCII_OTEL_METRIC_OUTBOUND_DROP, 1);
    }
    if (result == ASCIICHAT_OK) {
      atomic_fetch_add_u64(&out->submitted, 1);
      if (p->encoded && p->keyframe)
        atomic_fetch_add_u64(&out->recovery_frames, 1);
    }
    if (p->wait) {
      p->result = result;
      p->done = true;
      cond_broadcast(&out->changed);
    }
    if (result != ASCIICHAT_OK && !datagram_drop) {
      outbound_record_failure_and_fail_pending(out, result, &cause, &detail);
      break;
    }
  }
  bool failed = out->failed && !out->stopped_by_caller;
  mutex_unlock(&out->mutex);
  if (failed)
    out->transport->methods->close(out->transport);
  return NULL;
}
 
asciichat_error_t acip_outbound_create(acip_transport_t *transport) {
  ASCII_OTEL_SCOPE(ACIP_OUTBOUND_CREATE);
  if (!transport || !transport->methods || (!transport->methods->send && !transport->methods->sendv_ex))
    return SET_ERRNO(ERROR_INVALID_PARAM, "Missing outbound backend");
  if (transport->outbound)
    return ASCIICHAT_OK;
  acip_outbound_t *out = SAFE_CALLOC(1, sizeof(*out), acip_outbound_t *);
  if (mutex_init(&out->mutex, "connection_outgoing_packet_queue") != ASCIICHAT_OK) {
    SAFE_FREE(out);
    return GET_ERRNO();
  }
  if (cond_init(&out->changed, "outgoing_packet_ready_or_done") != ASCIICHAT_OK) {
    mutex_destroy(&out->mutex);
    SAFE_FREE(out);
    return GET_ERRNO();
  }
  out->transport = transport;
  asciichat_error_t result = platform_thread_create(&out->worker, "outbound", outbound_worker, out);
  if (result != ASCIICHAT_OK) {
    cond_destroy(&out->changed);
    mutex_destroy(&out->mutex);
    SAFE_FREE(out);
    return result;
  }
  transport->outbound = out;
  return ASCIICHAT_OK;
}
 
void acip_outbound_stop(acip_transport_t *transport) {
  acip_outbound_t *out = transport ? transport->outbound : NULL;
  if (!out)
    return;
  mutex_lock(&out->mutex);
  out->stopping = true;
  out->stopped_by_caller = true;
  outbound_record_failure_and_fail_pending(out, ERROR_NETWORK, NULL, NULL);
  mutex_unlock(&out->mutex);
}
 
void acip_outbound_destroy(acip_transport_t *transport) {
  ASCII_OTEL_SCOPE(ACIP_OUTBOUND_DESTROY);
  acip_outbound_t *out = transport->outbound;
  if (!out)
    return;
  mutex_lock(&out->mutex);
  out->stopping = true;
  cond_broadcast(&out->changed);
  mutex_unlock(&out->mutex);
  /* Wake a blocked send while every backend resource is still alive. The
   * failed-send worker also closes its backend, so wait for that worker before
   * closing here: two concurrent closes can join the same service thread. */
  if (transport->methods->interrupt)
    transport->methods->interrupt(transport);
  else
    transport->methods->close(transport);
  thread_join(&out->worker, NULL);
  if (transport->methods->interrupt)
    transport->methods->close(transport);
  mutex_lock(&out->mutex);
  outbound_record_failure_and_fail_pending(out, ERROR_NETWORK, NULL, NULL);
  while (out->recycled) {
    outbound_packet_t *p = out->recycled;
    out->recycled = p->next;
    packet_free(out, p);
  }
  mutex_unlock(&out->mutex);
  cond_destroy(&out->changed);
  mutex_destroy(&out->mutex);
  SAFE_FREE(transport->outbound);
}
 
static asciichat_error_t outbound_submitv(acip_transport_t *transport, const socket_buffer_t *buffers, size_t count,
                                          const acip_send_options_t *options, acip_admission_t *admission) {
  ASCII_OTEL_SCOPE(ACIP_OUTBOUND_SUBMITV);
  if (!transport || !transport->outbound || !buffers || !count || count > SOCKET_IOV_MAX || !admission)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid outbound submission");
  *admission = ACIP_ADMISSION_CONGESTED;
  size_t size = 0;
  for (size_t i = 0; i < count; ++i) {
    if ((buffers[i].len && !buffers[i].data) || buffers[i].len > ACIP_OUTBOUND_BUDGET - WIRE_OVERHEAD - size)
      return SET_ERRNO(ERROR_NETWORK_SIZE, "Outbound packet exceeds byte budget");
    size += buffers[i].len;
  }
  packet_header_t header;
  if (buffers[0].len < sizeof(header))
    return SET_ERRNO(ERROR_INVALID_PARAM, "Missing outbound ACIP header");
  memcpy(&header, buffers[0].data, sizeof(header));
  outbound_packet_t info = {.size = size,
                            .cost = 3 * (size + WIRE_OVERHEAD) + sizeof(outbound_packet_t) + 128,
                            .type = NET_TO_HOST_U16(header.type),
                            .stream = NET_TO_HOST_U32(header.client_id)};
  info.options = options ? *options
                         : (acip_send_options_t){.traffic_class = acip_packet_traffic_class(info.type),
                                                 .delivery = ACIP_DELIVERY_RELIABLE};
  if ((unsigned)info.options.traffic_class > ACIP_TRAFFIC_VISUAL ||
      (info.options.delivery != ACIP_DELIVERY_RELIABLE && info.options.delivery != ACIP_DELIVERY_DATAGRAM))
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid outbound send options");
  if (info.options.delivery != ACIP_DELIVERY_RELIABLE || info.options.deadline_ns) {
    if (!transport->methods->sendv_ex || !transport->methods->validate_send_options)
      return SET_ERRNO(ERROR_NOT_SUPPORTED, "Datagrams and deadlines are not supported by this transport");
    if (info.options.delivery == ACIP_DELIVERY_DATAGRAM &&
        (info.options.traffic_class == ACIP_TRAFFIC_CONTROL ||
         info.options.traffic_class != acip_packet_traffic_class(info.type)))
      return SET_ERRNO(ERROR_INVALID_PARAM, "Datagram flow does not match packet type");
    asciichat_error_t result = transport->methods->validate_send_options(transport, &info.options);
    if (result != ASCIICHAT_OK)
      return result;
    if (info.options.deadline_ns && info.options.deadline_ns <= time_get_ns())
      return SET_ERRNO(ERROR_NETWORK_TIMEOUT, "Outbound datagram deadline expired");
  }
  info.encoded = info.type == PACKET_TYPE_IMAGE_FRAME_H264 || info.type == PACKET_TYPE_IMAGE_FRAME_H265;
  info.video = info.encoded || info.type == PACKET_TYPE_IMAGE_FRAME || info.type == PACKET_TYPE_ASCII_FRAME;
  info.audio = acip_packet_traffic_class(info.type) == ACIP_TRAFFIC_AUDIO;
  ASCII_OTEL_ATTR_STRING("network.packet.category", info.video ? "video" : info.audio ? "audio" : "control");
  info.wait = !info.video && !info.audio && info.type != PACKET_TYPE_PING && info.type != PACKET_TYPE_PONG;
  if (info.encoded) {
    const uint8_t *flags = NULL;
    if (buffers[0].len > sizeof(header))
      flags = (const uint8_t *)buffers[0].data + sizeof(header);
    else if (count > 1 && buffers[1].len)
      flags = buffers[1].data;
    if (!flags)
      return SET_ERRNO(ERROR_INVALID_FRAME, "Missing encoded frame flags");
    info.keyframe = (*flags & 1) != 0;
  }
  acip_outbound_t *out = transport->outbound;
  mutex_lock(&out->mutex);
  info.created = time_get_ns();
  info.telemetry_context = ascii_otel_context();
  if (out->stopping || out->failed) {
    asciichat_error_t code = out->failed ? out->failure_code : ERROR_NETWORK;
    asciichat_error_context_t cause = out->failure;
    asciichat_error_context_t detail = out->failure_detail;
    mutex_unlock(&out->mutex);
    return cause.code == code ? import_failure(&cause, &detail) : SET_ERRNO(code, "Outbound connection closed");
  }
  // Admission first retires stale plaintext and restores codec recovery state;
  // dependent predictive frames stay rejected until a fresh keyframe arrives.
  expire_video(out, info.created);
  expire_datagrams(out, info.created);
  outbound_stream_t *stream = info.encoded ? stream_state(out, info.type, info.stream) : NULL;
  if (info.encoded && (!stream || (atomic_load_bool(&stream->needs_keyframe) && !info.keyframe))) {
    atomic_fetch_add_u64(&out->dropped, 1);
    ascii_otel_metric(ASCII_OTEL_METRIC_OUTBOUND_DROP, 1);
    *admission = ACIP_ADMISSION_NEEDS_KEYFRAME;
    ASCII_OTEL_ATTR_STRING("network.queue.admission", "needs_keyframe");
    mutex_unlock(&out->mutex);
    return ASCIICHAT_OK;
  }
  if (info.video && !info.encoded)
    discard_stream(out, &info);
  // Include backend-owned bytes before reserving room for audio and control.
  size_t backend = backend_bytes(out);
  size_t reserved = info.video   ? ACIP_OUTBOUND_CONTROL_RESERVE + ACIP_OUTBOUND_AUDIO_RESERVE
                    : info.audio ? ACIP_OUTBOUND_CONTROL_RESERVE
                                 : 0;
  size_t limit = ACIP_OUTBOUND_BUDGET - reserved;
  size_t pending = out->bytes + out->active_bytes + backend + transport->methods->reserved_bytes;
  ASCII_OTEL_ATTR_INT("network.queue.pending_bytes", (int64_t)pending);
  ASCII_OTEL_ATTR_INT("network.queue.budget_bytes", (int64_t)limit);
  ascii_otel_metric(ASCII_OTEL_METRIC_OUTBOUND_PENDING, (double)pending);
  if (out->head)
    ascii_otel_metric(ASCII_OTEL_METRIC_OUTBOUND_AGE, (double)(info.created - out->head->created) / 1e9);
  if (out->count >= OUTBOUND_MAX_PACKETS - (info.video   ? 64
                                            : info.audio ? 32
                                                         : 0) ||
      pending > limit || info.cost > limit - pending) {
    if (info.encoded)
      discard_stream(out, &info);
    atomic_fetch_add_u64(&out->congestion, 1);
    ASCII_OTEL_ATTR_STRING("network.queue.admission", "congested");
    ascii_otel_metric(ASCII_OTEL_METRIC_OUTBOUND_CONGESTION, 1);
    mutex_unlock(&out->mutex);
    return ASCIICHAT_OK;
  }
  // Reuse storage only when ownership and synchronous-wait semantics match.
  outbound_packet_t **cached = &out->recycled;
  while (*cached && ((*cached)->wait != info.wait || buffer_capacity((*cached)->data) < size))
    cached = &(*cached)->next;
  outbound_packet_t *p = *cached;
  uint8_t *storage = NULL;
  if (p) {
    *cached = p->next;
    --out->recycled_count;
    storage = p->data;
  } else {
    p = SAFE_MALLOC(sizeof(*p), outbound_packet_t *);
  }
  *p = info;
  p->data = storage ? storage : buffer_alloc(info.wait ? buffer_control_domain() : NULL, size);
  if (!p->data) {
    SAFE_FREE(p);
    mutex_unlock(&out->mutex);
    return SET_ERRNO(ERROR_MEMORY, "Cannot own outbound plaintext");
  }
  size_t offset = 0;
  for (size_t i = 0; i < count; ++i) {
    if (buffers[i].len)
      memcpy(p->data + offset, buffers[i].data, buffers[i].len);
    offset += buffers[i].len;
  }
  /* All non-video traffic is inserted ahead of pending media while preserving
   * protocol barriers and FIFO order within each traffic class. */
  bool was_empty = outbound_insert_packet(out, p, !info.video);
  out->bytes += info.cost;
  ++out->count;
  atomic_fetch_add_u64(&out->accepted, 1);
  pending += info.cost;
  uint64_t peak = atomic_load_u64(&out->peak_bytes);
  while (pending > peak && !atomic_cas_u64(&out->peak_bytes, &peak, pending))
    ;
  if (stream && info.keyframe)
    atomic_store_bool(&stream->needs_keyframe, false);
  *admission = ACIP_ADMISSION_ACCEPTED;
  ASCII_OTEL_ATTR_STRING("network.queue.admission", "accepted");
  if (was_empty)
    cond_broadcast(&out->changed);
  asciichat_error_t result = ASCIICHAT_OK;
  asciichat_error_context_t cause = {0};
  asciichat_error_context_t detail = {0};
  if (info.wait) {
    while (!p->done)
      cond_wait(&out->changed, &out->mutex);
    result = p->result;
    if (result != ASCIICHAT_OK) {
      cause = out->failure;
      detail = out->failure_detail;
    }
    packet_free(out, p);
  }
  mutex_unlock(&out->mutex);
  if (result == ASCIICHAT_OK)
    return result;
  return cause.code == result ? import_failure(&cause, &detail) : SET_ERRNO(result, "Outbound submission failed");
}
 
asciichat_error_t acip_outbound_submitv(acip_transport_t *transport, const socket_buffer_t *buffers, size_t count,
                                        acip_admission_t *admission) {
  return outbound_submitv(transport, buffers, count, NULL, admission);
}
 
asciichat_error_t acip_outbound_submitv_ex(acip_transport_t *transport, const socket_buffer_t *buffers, size_t count,
                                           const acip_send_options_t *options, acip_admission_t *admission) {
  if (!options)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Missing outbound send options");
  return outbound_submitv(transport, buffers, count, options, admission);
}
 
asciichat_error_t acip_outbound_snapshot(acip_transport_t *transport, acip_outbound_stats_t *stats) {
  if (!transport || !transport->outbound || !stats)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Missing outbound statistics");
  acip_outbound_t *out = transport->outbound;
  mutex_lock(&out->mutex);
  stats->accepted = atomic_load_u64(&out->accepted);
  stats->submitted = atomic_load_u64(&out->submitted);
  stats->dropped = atomic_load_u64(&out->dropped);
  stats->congestion = atomic_load_u64(&out->congestion);
  stats->recovery_frames = atomic_load_u64(&out->recovery_frames);
  stats->peak_bytes = (size_t)atomic_load_u64(&out->peak_bytes);
  stats->application_bytes = out->bytes + out->active_bytes;
  stats->backend_bytes = backend_bytes(out);
  stats->backend_reserved_bytes = transport->methods->reserved_bytes;
  uint64_t now = time_get_ns();
  for (outbound_packet_t *p = out->head; p; p = p->next)
    if (p->video && now - p->created > stats->oldest_frame_age_ns)
      stats->oldest_frame_age_ns = now - p->created;
  mutex_unlock(&out->mutex);
  return ASCIICHAT_OK;
}
 
bool acip_outbound_needs_keyframe(acip_transport_t *transport, uint16_t type, uint32_t id) {
  if (!transport || !transport->outbound)
    return false;
  acip_outbound_t *out = transport->outbound;
  if (transport->methods->consume_keyframe_request &&
      transport->methods->consume_keyframe_request(transport, type, id)) {
    outbound_packet_t stream = {.type = type, .stream = id, .video = true, .encoded = true};
    mutex_lock(&out->mutex);
    discard_stream(out, &stream);
    mutex_unlock(&out->mutex);
    return true;
  }
  mutex_lock(&out->mutex);
  size_t count = (size_t)atomic_load_u64(&out->stream_count);
  for (size_t i = 0; i < count; ++i)
    if (out->streams[i].type == type && out->streams[i].stream == id) {
      bool needs_keyframe = atomic_load_bool(&out->streams[i].needs_keyframe);
      mutex_unlock(&out->mutex);
      return needs_keyframe;
    }
  mutex_unlock(&out->mutex);
  return true;
}