endpoint.c

Go to the documentation of this file.

#include <ascii-chat/network/quic/quic.h>
#include <ascii-chat/common.h>
#include <ascii-chat/util/time.h>
#include "transport.h"
#include "worker.h"
 
#define QUIC_ENDPOINT_MAX_TRANSPORTS 64

typedef struct endpoint_transport {
  struct endpoint_transport *next;
  quic_connection_t *connection;
  acip_transport_t *transport;
  bool claimed;
} endpoint_transport_t;
 
struct quic_endpoint {
  mutex_t lane;
  quic_tls_t *tls;
  quic_udp_endpoint_t *udp;
  quic_worker_t *worker;
  endpoint_transport_t *transports;
  size_t count;
  bool listener, stopping, stopped, prefer_audio_datagrams, prefer_visual_datagrams;
  asciichat_error_t terminal_error;
};
 
static void wake_endpoint(void *context) {
  quic_endpoint_t *endpoint = context;
  (void)quic_udp_wake(endpoint->udp);
}
 
static void close_connection(quic_connection_t *connection, void *context) {
  quic_endpoint_t *endpoint = context;
  asciichat_errno_scope_t scope = errno_scope_begin();
  (void)quic_udp_close_peer(endpoint->udp, connection);
  errno_scope_end(scope, ASCIICHAT_ERRNO_DISMISSED);
}
 
/* The transport has already detached its receiver. Destroy may be called by
 * the normal network connection owner after an authenticated transport is adopted. */
static void transport_destroyed(acip_transport_t *transport, void *context) {
  quic_endpoint_t *endpoint = context;
  mutex_lock(&endpoint->lane);
  endpoint_transport_t **link = &endpoint->transports;
  while (*link && (*link)->transport != transport)
    link = &(*link)->next;
  endpoint_transport_t *item = *link;
  if (item) {
    if (item->connection && quic_udp_contains(endpoint->udp, item->connection)) {
      (void)quic_udp_remove(endpoint->udp, item->connection);
    }
    *link = item->next;
    --endpoint->count;
  }
  mutex_unlock(&endpoint->lane);
  SAFE_FREE(item);
}
 
/* Called on the endpoint lane. */
static asciichat_error_t attach(quic_endpoint_t *endpoint, quic_connection_t *connection, bool claimed,
                                acip_transport_t **out) {
  if (endpoint->stopping || endpoint->count >= QUIC_ENDPOINT_MAX_TRANSPORTS)
    return SET_ERRNO(ERROR_BUFFER_FULL, "QUIC endpoint transport limit reached");
  endpoint_transport_t *item = SAFE_CALLOC(1, sizeof(*item), endpoint_transport_t *);
  if (!item)
    return SET_ERRNO(ERROR_MEMORY, "Cannot allocate QUIC endpoint transport");
  asciichat_error_t result = quic_transport_attach_on_lane_with_waker(
      connection, &endpoint->lane, wake_endpoint, endpoint, transport_destroyed, endpoint, &item->transport);
  if (result != ASCIICHAT_OK) {
    SAFE_FREE(item);
    return result;
  }
  quic_transport_set_close_on_lane(item->transport, close_connection, endpoint);
  item->transport->prefer_audio_datagrams = endpoint->prefer_audio_datagrams;
  item->transport->prefer_visual_datagrams = endpoint->prefer_visual_datagrams;
  item->connection = connection;
  item->claimed = claimed;
  item->next = endpoint->transports;
  endpoint->transports = item;
  ++endpoint->count;
  if (out)
    *out = item->transport;
  return ASCIICHAT_OK;
}
 
static asciichat_error_t accept_connection(quic_connection_t *connection, void *context) {
  return attach(context, connection, false, NULL);
}
 
/* Called by the worker on the lane. Keep failed adapters alive until
 * their owners release them, but detach the connection before removing it. */
static bool connection_failed(quic_connection_t *connection, asciichat_error_t reason, void *context) {
  quic_endpoint_t *endpoint = context;
  if (!connection) {
    endpoint->terminal_error = reason;
    for (endpoint_transport_t *current = endpoint->transports; current; current = current->next) {
      (void)quic_transport_fail_on_lane(current->transport, reason);
      if (current->connection && quic_udp_remove(endpoint->udp, current->connection) == ASCIICHAT_OK)
        current->connection = NULL;
    }
    return false;
  }
  endpoint_transport_t *item = endpoint->transports;
  while (item && item->connection != connection)
    item = item->next;
  bool handled = !item && !quic_udp_contains(endpoint->udp, connection);
  if (item && quic_transport_fail_on_lane(item->transport, reason) == ASCIICHAT_OK &&
      quic_udp_remove(endpoint->udp, connection) == ASCIICHAT_OK) {
    item->connection = NULL;
    if (!item->claimed && quic_transport_destroy_unclaimed_on_lane(&item->transport, reason) == ASCIICHAT_OK) {
      endpoint_transport_t **link = &endpoint->transports;
      while (*link != item)
        link = &(*link)->next;
      *link = item->next;
      --endpoint->count;
      SAFE_FREE(item);
    }
    handled = true;
  }
  return handled;
}
 
asciichat_error_t quic_endpoint_open(const struct sockaddr *address, socklen_t length, bool listener,
                                     const quic_config_t *config, quic_endpoint_t **out) {
  if (!out)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Missing QUIC endpoint output");
  *out = NULL;
  if (!address || !config)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid QUIC endpoint configuration");
  quic_endpoint_t *endpoint = SAFE_CALLOC(1, sizeof(*endpoint), quic_endpoint_t *);
  if (!endpoint)
    return SET_ERRNO(ERROR_MEMORY, "Cannot allocate QUIC endpoint");
  if (mutex_init(&endpoint->lane, "quic_endpoint") != 0) {
    SAFE_FREE(endpoint);
    return SET_ERRNO(ERROR_THREAD, "Cannot initialize QUIC endpoint lane");
  }
  bool webtransport = config->webtransport_origin || config->webtransport_path;
  bool http3 = webtransport || config->enable_http3 || config->http_server;
  if (http3 && config->alpn && strcmp(config->alpn, "h3") != 0) {
    mutex_destroy(&endpoint->lane);
    SAFE_FREE(endpoint);
    return SET_ERRNO(ERROR_INVALID_PARAM, "HTTP/3 requires the h3 ALPN");
  }
  if (webtransport &&
      (!listener || !config->enable_datagrams || !config->webtransport_origin || !config->webtransport_path ||
       strncmp(config->webtransport_origin, "https://", sizeof("https://") - 1) != 0 ||
       config->webtransport_path[0] != '/' ||
       strnlen(config->webtransport_origin, sizeof(endpoint->tls->webtransport_origin)) >=
           sizeof(endpoint->tls->webtransport_origin) ||
       strnlen(config->webtransport_path, sizeof(endpoint->tls->webtransport_path)) >=
           sizeof(endpoint->tls->webtransport_path) ||
       (config->alpn && strcmp(config->alpn, "h3") != 0))) {
    mutex_destroy(&endpoint->lane);
    SAFE_FREE(endpoint);
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid WebTransport listener configuration");
  }
  const char *alpn = config->alpn ? config->alpn : http3 ? "h3" : "acip/1";
  asciichat_error_t result =
      listener ? quic_tls_server_create(alpn, config->certificate_file, config->private_key_file, &endpoint->tls)
               : quic_tls_client_create(alpn, config->ca_file, &endpoint->tls);
  if (result == ASCIICHAT_OK) {
    endpoint->tls->enable_datagrams = config->enable_datagrams;
    endpoint->tls->webtransport = webtransport;
    endpoint->tls->http3 = http3;
    endpoint->tls->accept_native = webtransport && config->accept_native;
    endpoint->tls->http_server = config->http_server;
    if (endpoint->tls->accept_native) {
      static const unsigned char protocols[] = {2, 'h', '3', 6, 'a', 'c', 'i', 'p', '/', '1'};
      memcpy(endpoint->tls->alpn, protocols, sizeof(protocols));
      endpoint->tls->alpn_length = sizeof(protocols);
    }
    if (webtransport) {
      memcpy(endpoint->tls->webtransport_origin, config->webtransport_origin,
             strlen(config->webtransport_origin) + (size_t)1);
      memcpy(endpoint->tls->webtransport_path, config->webtransport_path,
             strlen(config->webtransport_path) + (size_t)1);
    }
    endpoint->prefer_audio_datagrams = config->prefer_audio_datagrams;
    endpoint->prefer_visual_datagrams = config->prefer_visual_datagrams;
    endpoint->listener = listener;
    result = quic_udp_endpoint_create(address, length, listener ? endpoint->tls : NULL,
                                      listener ? accept_connection : NULL, listener ? endpoint : NULL, &endpoint->udp);
  }
  if (result == ASCIICHAT_OK)
    result = quic_worker_start_locked_failure(endpoint->udp, &endpoint->lane, connection_failed, endpoint,
                                              &endpoint->worker);
  if (result != ASCIICHAT_OK) {
    quic_udp_endpoint_destroy(&endpoint->udp);
    quic_tls_destroy(&endpoint->tls);
    mutex_destroy(&endpoint->lane);
    SAFE_FREE(endpoint);
    return result;
  }
  *out = endpoint;
  return ASCIICHAT_OK;
}
 
asciichat_error_t quic_endpoint_set_http_server(quic_endpoint_t *endpoint, http_server_t *server) {
  if (!endpoint || !endpoint->listener || !endpoint->tls || !endpoint->tls->http3)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Endpoint is not an HTTP/3 listener");
  mutex_lock(&endpoint->lane);
  asciichat_error_t result = ASCIICHAT_OK;
  if (endpoint->stopping || endpoint->count)
    result = SET_ERRNO(ERROR_INVALID_STATE, "HTTP/3 handler must be attached before connections");
  else
    endpoint->tls->http_server = server;
  mutex_unlock(&endpoint->lane);
  return result;
}
 
asciichat_error_t quic_endpoint_address(quic_endpoint_t *endpoint, struct sockaddr_storage *address,
                                        socklen_t *length) {
  if (!endpoint)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid QUIC endpoint");
  mutex_lock(&endpoint->lane);
  asciichat_error_t result = quic_udp_endpoint_address(endpoint->udp, address, length);
  mutex_unlock(&endpoint->lane);
  return result;
}
 
asciichat_error_t quic_connect(quic_endpoint_t *endpoint, const struct sockaddr *remote, socklen_t remote_length,
                               const char *server_name, acip_transport_t **out) {
  if (!out)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Missing QUIC transport output");
  *out = NULL;
  if (!endpoint || endpoint->listener || !remote || !server_name || !*server_name)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid QUIC connect configuration");
  mutex_lock(&endpoint->lane);
  quic_connection_t *connection = NULL;
  asciichat_error_t result = endpoint->stopping         ? SET_ERRNO(ERROR_INVALID_STATE, "QUIC endpoint stopped")
                             : endpoint->terminal_error ? SET_ERRNO(endpoint->terminal_error, "QUIC endpoint failed")
                             : endpoint->count >= QUIC_ENDPOINT_MAX_TRANSPORTS
                                 ? SET_ERRNO(ERROR_BUFFER_FULL, "QUIC endpoint transport limit reached")
                                 : quic_udp_connect(endpoint->udp, endpoint->tls, server_name, remote, remote_length,
                                                    time_get_ns(), &connection);
  if (result == ASCIICHAT_OK) {
    result = attach(endpoint, connection, true, out);
    if (result != ASCIICHAT_OK)
      (void)quic_udp_remove(endpoint->udp, connection);
  }
  mutex_unlock(&endpoint->lane);
  if (result == ASCIICHAT_OK)
    wake_endpoint(endpoint);
  return result;
}
 
asciichat_error_t quic_accept(quic_endpoint_t *endpoint, acip_transport_t **out) {
  if (!out)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Missing QUIC accept output");
  *out = NULL;
  if (!endpoint || !endpoint->listener)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid QUIC listener");
  mutex_lock(&endpoint->lane);
  endpoint_transport_t *item = endpoint->transports;
  bool generic_http3 = endpoint->tls->http3 && !endpoint->tls->webtransport;
  while (item && (item->claimed || generic_http3 ||
                  (endpoint->tls->webtransport && !quic_connection_accept_ready(item->connection))))
    item = item->next;
  asciichat_error_t result = ASCIICHAT_OK;
  if (item) {
    item->claimed = true;
    *out = item->transport;
  } else {
    result = endpoint->terminal_error ? SET_ERRNO(endpoint->terminal_error, "QUIC endpoint failed")
             : endpoint->stopping     ? SET_ERRNO(ERROR_INVALID_STATE, "QUIC endpoint stopped")
                                      : SET_ERRNO(ERROR_NETWORK_TIMEOUT, "No QUIC connection is available");
  }
  mutex_unlock(&endpoint->lane);
  return result;
}
 
asciichat_error_t quic_release(quic_endpoint_t *endpoint, acip_transport_t **transport) {
  if (!endpoint || !transport || !*transport)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid QUIC transport release");
  mutex_lock(&endpoint->lane);
  endpoint_transport_t **link = &endpoint->transports;
  while (*link && (*link)->transport != *transport)
    link = &(*link)->next;
  if (!*link || !(*link)->claimed) {
    mutex_unlock(&endpoint->lane);
    return SET_ERRNO(ERROR_INVALID_PARAM, "Transport does not belong to QUIC endpoint");
  }
  endpoint_transport_t *item = *link;
  asciichat_error_t result = quic_transport_fail_on_lane(item->transport, ERROR_NETWORK);
  if (result == ASCIICHAT_OK && item->connection)
    result = quic_udp_remove(endpoint->udp, item->connection);
  if (result == ASCIICHAT_OK) {
    *link = item->next;
    --endpoint->count;
  }
  mutex_unlock(&endpoint->lane);
  if (result != ASCIICHAT_OK)
    return result;
  acip_transport_destroy(item->transport);
  SAFE_FREE(item);
  *transport = NULL;
  return ASCIICHAT_OK;
}
 
asciichat_error_t quic_endpoint_stop(quic_endpoint_t *endpoint) {
  if (!endpoint)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid QUIC endpoint");
  if (endpoint->stopped)
    return ASCIICHAT_OK;
  mutex_lock(&endpoint->lane);
  endpoint->stopping = true;
  mutex_unlock(&endpoint->lane);
  asciichat_error_t worker_result = endpoint->worker ? quic_worker_stop(&endpoint->worker) : ASCIICHAT_OK;
  if (endpoint->worker)
    return worker_result;
  asciichat_error_t result = ASCIICHAT_OK;
  for (;;) {
    mutex_lock(&endpoint->lane);
    endpoint_transport_t *item = endpoint->transports;
    while (item && item->claimed)
      item = item->next;
    if (item)
      item->claimed = true;
    mutex_unlock(&endpoint->lane);
    if (!item)
      break;
    acip_transport_t *transport = item->transport;
    result = quic_release(endpoint, &transport);
    if (result != ASCIICHAT_OK)
      return result;
  }
  mutex_lock(&endpoint->lane);
  for (endpoint_transport_t *item = endpoint->transports; item; item = item->next) {
    result = quic_transport_fail_on_lane(item->transport, ERROR_NETWORK);
    if (result != ASCIICHAT_OK)
      break;
    if (item->connection) {
      result = quic_udp_remove(endpoint->udp, item->connection);
      if (result != ASCIICHAT_OK)
        break;
      item->connection = NULL;
    }
  }
  if (result == ASCIICHAT_OK)
    endpoint->stopped = true;
  mutex_unlock(&endpoint->lane);
  return result == ASCIICHAT_OK ? worker_result : result;
}
 
asciichat_error_t quic_endpoint_destroy(quic_endpoint_t **endpoint) {
  if (!endpoint || !*endpoint)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid QUIC endpoint destruction");
  if (!(*endpoint)->stopped || (*endpoint)->count)
    return SET_ERRNO(ERROR_INVALID_STATE, "Stop QUIC endpoint and release all transports first");
  quic_udp_endpoint_destroy(&(*endpoint)->udp);
  quic_tls_destroy(&(*endpoint)->tls);
  mutex_destroy(&(*endpoint)->lane);
  SAFE_FREE(*endpoint);
  *endpoint = NULL;
  return ASCIICHAT_OK;
}
 
asciichat_error_t quic_endpoint_migrate(quic_endpoint_t *endpoint, const struct sockaddr *local_address,
                                        socklen_t length) {
  if (!endpoint || endpoint->listener || !local_address)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid QUIC client endpoint migration");
  mutex_lock(&endpoint->lane);
  bool available = !endpoint->stopping && !endpoint->terminal_error && endpoint->count == 1 &&
                   endpoint->transports->connection && endpoint->worker;
  mutex_unlock(&endpoint->lane);
  if (!available)
    return SET_ERRNO(ERROR_INVALID_STATE, "QUIC migration requires one live client connection");
  asciichat_error_t result = quic_worker_stop(&endpoint->worker);
  if (result != ASCIICHAT_OK)
    return result;
  mutex_lock(&endpoint->lane);
  result = quic_udp_migrate(endpoint->udp, local_address, length, time_get_ns());
  asciichat_error_t restart =
      quic_worker_start_locked_failure(endpoint->udp, &endpoint->lane, connection_failed, endpoint, &endpoint->worker);
  if (restart != ASCIICHAT_OK)
    (void)connection_failed(NULL, restart, endpoint);
  mutex_unlock(&endpoint->lane);
  return restart != ASCIICHAT_OK ? restart : result;
}