video.cpp

/*
  The per-entry video child. See video.h for why it is independent of
  the MAVLink session.

  Phase 1 scope: process lifecycle, port binding, publisher admission
  and connections.tdb rows. There is no media path yet -- accepted
  bytes are counted and discarded -- so the process-model risk lands
  and can be tested on its own.
 */
#include "video.h"

#include <algorithm>
#include <initializer_list>
#include <memory>
#include <vector>

#include <errno.h>
#include <fcntl.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <time.h>
#include <unistd.h>

#include "conntdb.h"
#include "util.h"
#include "videoauth.h"
#include "videostream.h"
#include "videorec.h"
#include "videomkv.h"
#include "videortmp.h"
#include "videortsp.h"
#include "videoview.h"

#define VIDEO_MAX_EPOLL_EVENTS 64

// A publisher that goes quiet for this long releases its slot, so a
// replacement can take over after a genuine disconnect.
#define VIDEO_PUB_IDLE_S 10

/*
  Viewer table entries one source address may hold. Publishers are
  classified from a viewer slot, so without a cap a flood from one host
  fills the table and denies publishing outright. An operator watching
  three slots with a reconnect in flight needs a handful.
 */
#define VIDEO_MAX_VIEWERS_PER_IP 8

/*
  Concurrent RTMP handshakes per slot. More than one so a squatter
  cannot lock the camera out, few enough that the parser work an
  unauthenticated peer can start is bounded.
 */
#define VIDEO_MAX_PENDING_RTMP 4
// RTP datagrams to inspect for a decisive NAL type before assuming H.264
#define VIDEO_RTP_SNIFF_MAX 64
// wait this long after a failed backend start before trying again
#define VIDEO_RTP_RETRY_S 5

// How often the child re-reads its own keys.tdb record and re-writes
// its connections.tdb rows. Matches the MAVLink child's cadence.
#define VIDEO_TICK_S 5

// Rate limit for "rejected" logging, so a flood of unauthorised packets
// can't turn into a flood of stdout writes (which would block the child
// once the pipe fills).
#define VIDEO_LOG_MIN_INTERVAL_S 1

bool video_entry_wants_child(uint32_t flags, const uint32_t *video_ports)
{
    if ((flags & KEY_FLAG_VIDEO) == 0) {
        return false;
    }
    for (int i = 0; i < KEY_MAX_VIDEO_PORTS; i++) {
        if (video_ports[i] != 0) {
            return true;
        }
    }
    return false;
}

namespace {

/*
  One direction of the splice.

  The relay used to retry a short write by sleeping inside the event
  loop. That loop is the only thread: it also drains ffmpeg's stdout, so
  sleeping there deadlocks -- the publisher fills the socket to ffmpeg,
  we stop servicing epoll, ffmpeg's stdout pipe fills, ffmpeg stops
  reading its input, and the socket never drains. A paced publisher
  (ffmpeg -re, as the tests use) never triggers it; a real camera bursts.
 */
struct SpliceQueue {
    std::vector<uint8_t> buf;
    size_t sent = 0;
    bool armed = false;          // EPOLLOUT currently armed on the sink

    size_t pending(void) const { return buf.size() - sent; }
    void clear(void) { buf.clear(); sent = 0; armed = false; }
    void compact(void)
    {
        if (sent == buf.size()) {
            buf.clear();
            sent = 0;
        } else if (sent > 65536) {
            buf.erase(buf.begin(), buf.begin() + long(sent));
            sent = 0;
        }
    }
};

// Stop reading a direction once this much is already queued for it, so
// the backlog is bounded and TCP applies the backpressure upstream
// instead of this process buffering without limit.
#define SPLICE_QUEUE_MAX (1u * 1024 * 1024)

/*
  RTSP is relayed byte-for-byte, but admission credentials can occur on
  any request URI in the session. Hold only the current request header
  long enough to inspect its request line. Bodies and interleaved RTP are
  counted and streamed without interpretation.
 */
struct RtspRequestGuard {
    std::vector<uint8_t> buffered;
    size_t opaque_left = 0;

    void clear(void) { buffered.clear(); opaque_left = 0; }
};

/*
  One RTMP handshake that has not published yet.

  It owns nothing but its socket: no slot, no backend, no ring. Only
  when publish arrives and passes admission is one of these promoted.
 */
struct PendingRtmp {
    int fd = -1;
    std::unique_ptr<RtmpSession> sess;
    SpliceQueue out;            // responses owed to this peer
    time_t since = 0;
    uint32_t ip_be = 0;
    uint16_t port_be = 0;

    bool active(void) const { return fd >= 0; }
};

struct Slot {
    uint32_t port = 0;
    int udp_fd = -1;
    int tcp_fd = -1;

    // current publisher, if any
    bool has_pub = false;
    uint32_t pub_ip_be = 0;
    uint16_t pub_port_be = 0;
    time_t pub_since = 0;
    time_t pub_last = 0;
    uint64_t pub_bytes = 0;

    // rate-limited rejection logging
    time_t last_reject_log = 0;
    uint32_t rejects = 0;

    // RTSP or RTMP publisher, while one holds the slot
    RtspBackend rtsp;
    int rtsp_client_fd = -1;
    SpliceQueue to_backend;   // bytes read from the client, owed to ffmpeg
    SpliceQueue to_client;    // and the other way
    RtspRequestGuard rtsp_guard;
    /*
      The RTMP session that owns the slot, once one has published and
      been admitted. Null until then.
     */
    std::unique_ptr<RtmpSession> rtmp;

    /*
      Handshakes in progress. These deliberately do NOT hold the slot.

      Classification costs one byte, so if a pending handshake owned the
      slot an unauthenticated peer could take it, wait out its deadline
      and reconnect, denying publishing indefinitely -- and letting a
      newcomer evict the incumbent only turns that into last-arrival
      wins, which is the same denial. Several negotiate side by side
      instead, and the slot is awarded on publish, after admission.
     */
    PendingRtmp pending[VIDEO_MAX_PENDING_RTMP];

    int mkv_fd = -1;
    VideoChunks chunks;

    // media
    VideoRing ring;
    VideoScanner scanner;
    VideoWriter rec;
    VideoViewer viewers[VIDEO_MAX_VIEWERS];
    uint32_t viewer_events[VIDEO_MAX_VIEWERS] {};   // last epoll mask armed
    uint32_t viewers_seen = 0;
    uint32_t viewers_dropped = 0;
    bool recording = false;
    uint64_t last_anchor = 0;
    bool had_anchor = false;
    uint64_t bad_datagrams = 0;
    bool warned_204 = false;

    // bare RTP over UDP: datagrams seen while deciding the codec
    uint32_t rtp_seen = 0;
    time_t rtp_retry_after = 0;
    void reset_rtp(void) {
        rtp_seen = 0;
        rtp_retry_after = 0;
    }
};

enum rtp_codec_t {
    RTP_CODEC_UNKNOWN = 0,
    RTP_CODEC_H264,
    RTP_CODEC_HEVC,
};

/*
  Offset of the RTP payload, or -1 if this is not an RTP datagram we
  would forward: version 2, a dynamic payload type (which also excludes
  RTCP, whose types land at 72..76 after the marker bit is masked).
 */
static int rtp_payload_offset(const uint8_t *buf, size_t n)
{
    if (n < 13 || (buf[0] & 0xC0) != 0x80) {
        return -1;
    }
    const int pt = buf[1] & 0x7f;
    if (pt < 96 || pt > 127) {
        return -1;
    }
    size_t off = 12 + 4u * (buf[0] & 0x0f);
    if (buf[0] & 0x10) {                     // extension header
        if (off + 4 > n) {
            return -1;
        }
        off += 4 + 4u * ((size_t(buf[off + 2]) << 8) | buf[off + 3]);
    }
    return off < n ? int(off) : -1;
}

/*
  Decide the codec from the NAL header. Only values that separate the
  two layouts count: H.264 FU-A, STAP-A, IDR and nri=3 SPS/PPS map to
  reserved HEVC types; HEVC FU, AP and parameter sets with a layer id
  of 0 map to H.264 types 0/2/4, which are unspecified or the
  Extended-profile data partitions no camera emits. Everything else is
  a single slice that could be either, so the caller keeps looking.
 */
static rtp_codec_t rtp_codec_hint(const uint8_t *p, size_t n)
{
    if (n < 2 || (p[0] & 0x80) != 0) {
        return RTP_CODEC_UNKNOWN;
    }
    switch (p[0]) {
    case 0x40: case 0x42: case 0x44:     // VPS, SPS, PPS
    case 0x60: case 0x62:                // AP, FU
        // second byte: layer id low bits (must be 0) and TID+1 (>= 1)
        return (p[1] & 0xF8) == 0 && (p[1] & 0x07) != 0
            ? RTP_CODEC_HEVC : RTP_CODEC_UNKNOWN;
    default:
        break;
    }
    switch (p[0] & 0x1f) {
    case 28: case 24:                    // FU-A, STAP-A
        return RTP_CODEC_H264;
    case 5: case 7: case 8:              // IDR, SPS, PPS
        return (p[0] & 0x60) == 0x60 ? RTP_CODEC_H264 : RTP_CODEC_UNKNOWN;
    default:
        return RTP_CODEC_UNKNOWN;
    }
}

class VideoChild {
public:
    VideoChild(int port2, int ready_fd) :
        port2_(port2), ready_fd_(ready_fd), auth_(port2) {}

    void run(void) __attribute__((noreturn));

private:
    int port2_;
    int ready_fd_;
    VideoAuth auth_;
    struct KeyEntry ke_ {};
    Slot slots_[KEY_MAX_VIDEO_PORTS];
    int epfd_ = -1;
    time_t last_tick_ = 0;

    bool load_entry(void);
    int bind_slots(void);
    void signal_ready(int err);
    void handle_mkv(Slot &s, int idx, int fd, struct sockaddr_in &from, time_t now);
    bool pump_mkv(Slot &s, int idx, time_t now);
    void close_mkv(Slot &s, int idx, const char *why);
    void handle_udp(Slot &s, int idx);
    void handle_tcp(Slot &s, int idx);
    void tick(time_t now);
    void write_conn_rows(time_t now);
    // Does this slot fall back to the MAVLink session when a publish
    // password is set but the publisher offered none?
    bool session_ok(int idx) const {
        return (video_slot_opts_of(ke_, unsigned(idx))
                & VIDEO_SLOT_SESSION_OK) != 0;
    }
    // Does this slot admit a publisher that offered no credential with
    // no MAVLink-session check either?
    bool open_publish(int idx) const {
        return (video_slot_opts_of(ke_, unsigned(idx))
                & VIDEO_SLOT_OPEN_PUB) != 0;
    }

    void log_reject(Slot &s, int idx, uint32_t ip_be, video_admit_t r,
                    time_t now);
    void ingest_udp(Slot &s, int idx, const uint8_t *buf, size_t n,
                    time_t now);
    void ingest_rtp(Slot &s, int idx, const uint8_t *buf, size_t n,
                    time_t now);
    void ingest(Slot &s, int idx, const uint8_t *buf, size_t n);
    void ingest_stream(Slot &s, int idx, const uint8_t *buf, size_t n);
    void handle_rtsp(Slot &s, int idx, int fd,
                     struct sockaddr_in &from, time_t now,
                     splice_proto_t proto);
    void close_rtsp(Slot &s, int idx, const char *why);
    bool pump_rtsp(Slot &s, int idx, int fd, time_t now);
    bool guard_rtsp_requests(Slot &s, int idx, const uint8_t *buf, size_t n,
                             time_t now);
    bool pump_rtmp(Slot &s, int idx, int fd, time_t now);
    bool rtmp_start_backend(Slot &s, int idx);
    bool rtmp_drain_owner(Slot &s, int idx, bool alive);
    void close_pending(Slot &s, int idx, PendingRtmp &p, const char *why);
    bool pump_pending(Slot &s, int idx, PendingRtmp &p, time_t now);
    bool promote_pending(Slot &s, int idx, PendingRtmp &p, time_t now);
    void latch_publisher(Slot &s, int idx, time_t now);
    bool splice_flush(int to_fd, SpliceQueue &q);
    void splice_arm(int to_fd, SpliceQueue &q);
    void epoll_add_viewer(VideoViewer &v, uint32_t &events);
    void epoll_sync_viewer(VideoViewer &v, uint32_t &events);
    void drop_viewer(Slot &s, int idx, VideoViewer &v);
    void end_stream(Slot &s, int idx, const char *why);
    void pump_viewers(Slot &s, int idx, time_t now);
};

bool VideoChild::load_entry(void)
{
    auto *db = db_open();
    if (db == nullptr) {
        return false;
    }
    struct KeyEntry k {};
    bool ok = db_load_key(db, port2_, k);
    db_close(db);
    if (ok) {
        ke_ = k;
    }
    return ok;
}

void VideoChild::signal_ready(int err)
{
    if (ready_fd_ < 0) {
        return;
    }
    uint8_t b = uint8_t(err > 255 ? 255 : err);
    ssize_t n = ::write(ready_fd_, &b, 1);
    (void)n;
    close(ready_fd_);
    ready_fd_ = -1;
}

int VideoChild::bind_slots(void)
{
    int first_err = 0;
    for (int i = 0; i < KEY_MAX_VIDEO_PORTS; i++) {
        const uint32_t port = video_port_of(ke_, unsigned(i));
        if (port == 0) {
            continue;
        }
        slots_[i].port = port;
        slots_[i].udp_fd = open_socket_in_udp(int(port));
        if (slots_[i].udp_fd == -1 && first_err == 0) {
            first_err = errno ? errno : EADDRINUSE;
        }
        slots_[i].tcp_fd = open_socket_in_tcp(int(port));
        if (slots_[i].tcp_fd == -1 && first_err == 0) {
            first_err = errno ? errno : EADDRINUSE;
        }
        if (slots_[i].udp_fd != -1 || slots_[i].tcp_fd != -1) {
            printf("[%d] video slot %d listening on %u%s\n",
                   port2_, i, unsigned(port),
                   (slots_[i].udp_fd == -1 || slots_[i].tcp_fd == -1)
                   ? " (partially)" : "");
        } else {
            printf("[%d] video slot %d failed to bind %u - %s\n",
                   port2_, i, unsigned(port), strerror(errno));
        }
    }
    return first_err;
}

void VideoChild::log_reject(Slot &s, int idx, uint32_t ip_be,
                            video_admit_t r, time_t now)
{
    s.rejects++;
    if (now - s.last_reject_log < VIDEO_LOG_MIN_INTERVAL_S) {
        return;
    }
    s.last_reject_log = now;
    struct in_addr a {};
    a.s_addr = ip_be;
    printf("[%d] video slot %d rejected %s: %s (%u so far)\n",
           port2_, idx, inet_ntoa(a), video_admit_str(r), unsigned(s.rejects));
}

/*
  Accept one datagram's worth of publisher bytes.

  A datagram must be a whole number of 188-byte TS packets starting with
  the sync byte -- that is what every MPEG-TS/UDP sender produces (7
  packets, 1316 bytes, is the norm). Anything else is dropped and
  counted rather than fed to the scanner, so a misconfigured sender
  shows up as a clear count instead of a stream that half works.
 */
void VideoChild::ingest(Slot &s, int idx, const uint8_t *buf, size_t n)
{
    if (n < TS_PACKET_SIZE || buf[0] != TS_SYNC_BYTE
        || (n % TS_PACKET_SIZE) != 0
        || (n > TS_PACKET_SIZE && buf[TS_PACKET_SIZE] != TS_SYNC_BYTE)) {
        // 204-byte packets are DVB's TS-with-Reed-Solomon. Feeding them
        // to a 188-byte parser produces nonsense, so say so once.
        if (!s.warned_204 && (n % 204) == 0 && n >= 204
            && buf[0] == TS_SYNC_BYTE) {
            s.warned_204 = true;
            printf("[%d] video slot %d: 204-byte (DVB) packets are not "
                   "supported; send 188-byte MPEG-TS\n", port2_, idx);
        }
        s.bad_datagrams++;
        return;
    }
    s.pub_bytes += n;
    // Scanner first: it needs the offset this data will occupy, and the
    // ring write is what makes that offset meaningful.
    const uint64_t before = s.ring.write_pos();
    s.scanner.feed(buf, n, before);
    s.ring.write(buf, n);

    if (!s.recording || s.rec.stopped()) {
        return;
    }
    const time_t now = time(nullptr);

    /*
      Cut segments at a join boundary so every segment is playable from
      its first byte. The anchor only moves when a new random access
      point arrives, so "the anchor advanced into this datagram" is the
      signal that we are standing on one.

      A stream whose muxer never signals one must still rotate, hence
      the overshoot fallback -- otherwise the segment grows forever and
      the quota pass can never evict it.
     */
    if (s.rec.rotation_due(now)) {
        uint64_t anchor = 0;
        const bool have = s.scanner.join_offset(anchor);
        const bool at_boundary = have && s.had_anchor && anchor > s.last_anchor
            && anchor >= before;
        if (at_boundary || s.rec.rotation_overdue(now)) {
            s.rec.rotate(now, at_boundary);
        }
    }
    uint64_t anchor_now = 0;
    if (s.scanner.join_offset(anchor_now)) {
        s.last_anchor = anchor_now;
        s.had_anchor = true;
    }
    s.rec.write(buf, n, now);
}

/*
  Byte-stream ingest, for a source that is not datagram-framed (the
  RTSP backend's stdout). The 188-alignment rule that guards the UDP
  path does not apply -- packets straddle reads by nature -- and the
  scanner and ring already carry partial packets across calls.
 */
void VideoChild::ingest_stream(Slot &s, int idx, const uint8_t *buf, size_t n)
{
    s.pub_bytes += n;
    const uint64_t before = s.ring.write_pos();
    s.scanner.feed(buf, n, before);
    s.ring.write(buf, n);

    if (!s.recording || s.rec.stopped()) {
        return;
    }
    const time_t now = time(nullptr);
    if (s.rec.rotation_due(now)) {
        uint64_t anchor = 0;
        const bool have = s.scanner.join_offset(anchor);
        const bool at_boundary = have && s.had_anchor && anchor > s.last_anchor
            && anchor >= before;
        if (at_boundary || s.rec.rotation_overdue(now)) {
            s.rec.rotate(now, at_boundary);
        }
    }
    uint64_t anchor_now = 0;
    if (s.scanner.join_offset(anchor_now)) {
        s.last_anchor = anchor_now;
        s.had_anchor = true;
    }
    s.rec.write(buf, n, now);
    (void)idx;
}

void VideoChild::handle_udp(Slot &s, int idx)
{
    uint8_t buf[65536];    // the largest IPv4 datagram fits
    struct sockaddr_in from {};
    socklen_t fromlen = sizeof(from);
    ssize_t n = recvfrom(s.udp_fd, buf, sizeof(buf), MSG_TRUNC,
                         (struct sockaddr *)&from, &fromlen);
    if (n <= 0) {
        return;
    }
    if (s.mkv_fd >= 0) return; // this slot belongs to the authenticated TCP publisher
    if (size_t(n) > sizeof(buf)) {
        // MSG_TRUNC reports the real length: a truncated RTP packet
        // would be forwarded as a whole one, so drop it instead.
        s.bad_datagrams++;
        return;
    }
    const time_t now = time(nullptr);

    // A datagram from the established publisher needs no re-check: the
    // admission decision was made when the tuple latched.
    if (s.has_pub && s.pub_ip_be == uint32_t(from.sin_addr.s_addr)
        && s.pub_port_be == from.sin_port) {
        s.pub_last = now;
        ingest_udp(s, idx, buf, size_t(n), now);
        return;
    }

    // Plain MPEG-TS over UDP carries no credential, so path A can't
    // apply here; admit() falls through to the MAVLink-session check
    // unless a publish password is set and the slot is not flagged
    // session_ok, in which case UDP can't satisfy it and the datagram
    // is refused.
    video_admit_t r = auth_.admit(ke_, uint32_t(from.sin_addr.s_addr),
                                  nullptr, false, session_ok(idx),
                                  open_publish(idx), now);
    if (r != VIDEO_ADMIT_OK) {
        log_reject(s, idx, uint32_t(from.sin_addr.s_addr), r, now);
        return;
    }
    if ((s.has_pub && now - s.pub_last <= VIDEO_PUB_IDLE_S)
        || s.rtsp.running() || s.rtsp_client_fd >= 0) {
        // One publisher at a time: a second sender behind the same NAT
        // must not be able to interleave into the stream. This is not
        // an authorisation failure -- the sender may be perfectly
        // entitled to publish -- so it gets its own reason rather than
        // borrowing one that would send an operator hunting an address
        // mismatch that isn't there.
        //
        // The connection tests are part of it, not just has_pub: a
        // spliced or RTMP publisher owns the slot for as long as its
        // connection lives, and testing has_pub alone let a UDP sender
        // latch on top of a live one and mix two streams into one ring.
        // rtsp_client_fd covers an RTMP handshake still in progress,
        // which has no backend and no latched publisher yet.
        log_reject(s, idx, uint32_t(from.sin_addr.s_addr),
                   VIDEO_ADMIT_SLOT_BUSY, now);
        return;
    }
    s.pub_ip_be = uint32_t(from.sin_addr.s_addr);
    s.pub_port_be = from.sin_port;
    latch_publisher(s, idx, now);
    printf("[%d] video slot %d publisher %s\n",
           port2_, idx, addr_to_str(from));
    ingest_udp(s, idx, buf, size_t(n), now);
}

/*
  A datagram from the latched UDP publisher: MPEG-TS goes straight to
  the scanner; bare RTP goes to a backend that muxes it, started once
  the codec is known.
 */
void VideoChild::ingest_udp(Slot &s, int idx, const uint8_t *buf, size_t n,
                            time_t now)
{
    if (s.rtsp.running()) {
        if (s.rtsp.proto() == SPLICE_RTP) {
            s.rtsp.send_rtp(buf, n);
        }
        return;
    }
    // Only a stream that has not yet delivered any TS can be RTP: once
    // packets have been scanned, an RTP-shaped datagram is a stray, not
    // a reason to abandon the stream.
    if (s.rtp_seen > 0 || (s.scanner.stats().packets == 0
                           && rtp_payload_offset(buf, n) >= 0)) {
        ingest_rtp(s, idx, buf, n, now);
        return;
    }
    ingest(s, idx, buf, n);
}

void VideoChild::ingest_rtp(Slot &s, int idx, const uint8_t *buf, size_t n,
                            time_t now)
{
    const int off = rtp_payload_offset(buf, n);
    if (off < 0) {
        return;
    }
    if (s.rtp_seen < UINT32_MAX) {
        s.rtp_seen++;
    }
    if (now < s.rtp_retry_after) {
        return;
    }
    rtp_codec_t codec = rtp_codec_hint(buf + off, n - size_t(off));
    if (codec == RTP_CODEC_UNKNOWN) {
        if (s.rtp_seen < VIDEO_RTP_SNIFF_MAX) {
            return;
        }
        codec = RTP_CODEC_H264;
    }
    const char *name = codec == RTP_CODEC_HEVC ? "H265" : "H264";
    if (!s.rtsp.start(port2_, idx, false, SPLICE_RTP, nullptr, name,
                      buf[1] & 0x7f)) {
        s.rtp_retry_after = now + VIDEO_RTP_RETRY_S;
        return;
    }
    if (s.rtsp.media_fd() >= 0) {
        struct epoll_event ev {};
        ev.events = EPOLLIN | EPOLLRDHUP;
        ev.data.fd = s.rtsp.media_fd();
        epoll_ctl(epfd_, EPOLL_CTL_ADD, s.rtsp.media_fd(), &ev);
    }
    s.rtsp.send_rtp(buf, n);
}

void VideoChild::handle_tcp(Slot &s, int idx)
{
    struct sockaddr_in from {};
    socklen_t fromlen = sizeof(from);
    int fd = accept(s.tcp_fd, (struct sockaddr *)&from, &fromlen);
    if (fd == -1) {
        return;
    }
    const time_t now = time(nullptr);

    /*
      A TCP connection here is a viewer. Viewers are not publishers and
      are not subject to the publisher admission check: they present a
      viewer password (or the slot is open), and in either case they can
      only ever reach a stream that an authorised publisher is feeding.
     */
    /*
      A publisher arrives through this table too -- RTSP and RTMP are
      only recognised once bytes arrive, so the connection has to be
      held somewhere first. That means a flood of connections can fill
      the table and stop a publisher from even being classified.

      Cap how many one address may hold. It cannot be a reserve for
      "probable publishers": a publisher is indistinguishable at accept
      time, since the credential it might carry arrives later. A per
      address cap needs none of that, and one host can no longer take
      the table on its own.
     */
    int free_slot = -1;
    int from_this_ip = 0;
    for (int v = 0; v < VIDEO_MAX_VIEWERS; v++) {
        if (s.viewers[v].active()
            && s.viewers[v].peer_ip_be() == uint32_t(from.sin_addr.s_addr)) {
            from_this_ip++;
        }
    }
    if (from_this_ip < VIDEO_MAX_VIEWERS_PER_IP) {
        for (int v = 0; v < VIDEO_MAX_VIEWERS; v++) {
            if (!s.viewers[v].active()) {
                free_slot = v;
                break;
            }
        }
    }
    if (free_slot < 0) {
        // Say why rather than dropping silently; a viewer that just
        // disconnects with no explanation is impossible to diagnose.
        const std::string body = http_simple_response(
            503, "too many viewers", "text/plain",
            "This stream already has the maximum number of viewers.\n");
        (void)::send(fd, body.data(), body.size(), MSG_NOSIGNAL);
        close(fd);
        printf("[%d] video slot %d viewer from %s refused: slot full\n",
               port2_, idx, addr_to_str(from));
        return;
    }
    s.viewers[free_slot].start(fd, port2_, uint32_t(from.sin_addr.s_addr),
                               from.sin_port, now);
    s.viewer_events[free_slot] = 0;
    s.viewers_seen++;
    epoll_add_viewer(s.viewers[free_slot], s.viewer_events[free_slot]);
    printf("[%d] video slot %d viewer from %s connected\n",
           port2_, idx, addr_to_str(from));
}

/*
  Take a slot for a publisher whose media arrives on the backend's
  stdout rather than as datagrams. Everything downstream -- the ring,
  the scanner, the recorder, the ConnEntry row -- is the same.
 */
void VideoChild::latch_publisher(Slot &s, int idx, time_t now)
{
    s.has_pub = true;
    s.pub_since = now;
    s.pub_last = now;
    s.pub_bytes = 0;
    s.ring.init(video_ring_bytes());
    s.scanner = VideoScanner();
    s.rec.set_matroska(false);
    s.had_anchor = false;
    s.reset_rtp();
    s.recording = (video_slot_opts_of(ke_, unsigned(idx))
                   & VIDEO_SLOT_RECORD) != 0;
    if (s.recording) {
        s.rec.configure(uint32_t(port2_), idx,
                        (ke_.flags & KEY_FLAG_USE_TZ) != 0,
                        ke_.tz_offset_hours, "logs", ke_.video_quota_mb);
    }
    last_tick_ = 0;   // snapshot connections.tdb promptly
}

void VideoChild::handle_mkv(Slot &s, int idx, int fd,
                            struct sockaddr_in &from, time_t now)
{
    uint8_t buf[2048];
    ssize_t n = recv(fd, buf, sizeof(buf), MSG_PEEK);
    HttpRequest req;
    if (n <= 0 || req.feed(buf, size_t(n)) != 1) { close(fd); return; }
    char path[32]; snprintf(path, sizeof(path), "/v%d.mkv", idx+1);
    int code = 400;
    std::string pw;
    bool offered = http_query_value(req.target(), "pw", pw);
    auto admission = auth_.admit(ke_, uint32_t(from.sin_addr.s_addr), offered ? &pw : nullptr,
                                 true, session_ok(idx), open_publish(idx), now);
    bool valid = req.path() == path && req.header_count("Content-Type") == 1 &&
        req.header("Content-Type") == "video/x-matroska" &&
        req.header_count("Transfer-Encoding") == 1 && req.header("Transfer-Encoding") == "chunked" &&
        req.header_count("Content-Length") == 0 && req.header("Expect") == "100-continue";
    if (admission != VIDEO_ADMIT_OK) code = 403;
    else if (s.has_pub || s.rtsp.running() || s.mkv_fd >= 0) code = 409;
    else if (valid) code = 100;
    if (code != 100) {
        const auto response = http_simple_response(code, "publish rejected", "text/plain", "Cannot publish this stream.\n");
        (void)send(fd, response.data(), response.size(), MSG_NOSIGNAL);
        close(fd); return;
    }
    const size_t header_size = size_t(n)-req.leftover().size();
    if (recv(fd, buf, header_size, 0) != ssize_t(header_size)) { close(fd); return; }
    const char response[] = "HTTP/1.1 100 Continue\r\n\r\n";
    if (send(fd, response, sizeof(response)-1, MSG_NOSIGNAL) != sizeof(response)-1) { close(fd); return; }
    s.pub_ip_be = from.sin_addr.s_addr; s.pub_port_be = from.sin_port;
    latch_publisher(s, idx, now);
    s.mkv_fd = fd; s.chunks = VideoChunks(); s.scanner.matroska = true;
    s.rec.set_matroska(true);
    s.scanner.cluster = [&s](const uint8_t *data, size_t length) {
        if (!s.recording || s.rec.stopped()) return;
        const time_t t = time(nullptr);
        if (s.rec.rotation_due(t)) s.rec.rotate(t, true);
        if (!s.rec.is_open()) {
            const auto &header = s.scanner.prefix;
            if (!s.rec.write(reinterpret_cast<const uint8_t *>(header.data()), header.size(), t)) return;
        }
        s.rec.write(data, length, t);
    };
    struct epoll_event ev {}; ev.events = EPOLLIN | EPOLLRDHUP; ev.data.fd = fd;
    epoll_ctl(epfd_, EPOLL_CTL_ADD, fd, &ev);
    printf("[%d] video slot %d Matroska publisher connected\n", port2_, idx);
}

bool VideoChild::pump_mkv(Slot &s, int idx, time_t now)
{
    uint8_t buf[65536];
    const ssize_t n = recv(s.mkv_fd, buf, sizeof(buf), 0);
    if (n < 0) return errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR;
    if (!n) return false;
    s.pub_last = now;
    bool ok = s.chunks.feed(buf, size_t(n), [&s](const uint8_t *data, size_t length) {
        s.scanner.feed(data, length, s.ring.write_pos());
        s.ring.write(data, length); s.pub_bytes += length;
    });
    (void)idx;
    return ok && !s.scanner.failed;
}

void VideoChild::close_mkv(Slot &s, int idx, const char *why)
{
    epoll_ctl(epfd_, EPOLL_CTL_DEL, s.mkv_fd, nullptr);
    close(s.mkv_fd); s.mkv_fd = -1; s.has_pub = false;
    s.rec.close_segment(); s.recording = false;
    end_stream(s, idx, why);
}

void VideoChild::handle_rtsp(Slot &s, int idx, int fd,
                             struct sockaddr_in &from, time_t now,
                             splice_proto_t proto)
{
    /*
      RTMP is not spliced: we speak it ourselves, and the credential
      only arrives with publish. So the connection is parked in the
      pending pool, which owns no part of the slot, and admission
      happens later.
     */
    if (proto == SPLICE_RTMP) {
        int free_i = -1;
        int oldest_i = -1;
        for (int i = 0; i < VIDEO_MAX_PENDING_RTMP; i++) {
            if (!s.pending[i].active()) {
                free_i = i;
                break;
            }
            if (oldest_i < 0 || s.pending[i].since < s.pending[oldest_i].since) {
                oldest_i = i;
            }
        }
        if (free_i < 0) {
            // Pool full: drop the oldest, which has had the longest to
            // publish and has not. Bounded either way, and a squatter
            // cannot pin every entry against a camera that retries.
            close_pending(s, idx, s.pending[oldest_i], "pending pool full");
            free_i = oldest_i;
        }
        PendingRtmp &p = s.pending[free_i];
        fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
        p.fd = fd;
        p.sess.reset(new RtmpSession());
        p.out.clear();
        p.since = now;
        p.ip_be = uint32_t(from.sin_addr.s_addr);
        p.port_be = from.sin_port;
        printf("[%d] video slot %d RTMP publisher %s connecting\n",
               port2_, idx, addr_to_str(from));
        struct epoll_event ev {};
        ev.events = EPOLLIN | EPOLLRDHUP;
        ev.data.fd = fd;
        epoll_ctl(epfd_, EPOLL_CTL_ADD, fd, &ev);
        return;
    }

    /*
      Read a publish password out of the request-line URI, if there is
      one, e.g. rtsp://host:port/cam?pw=secret. Peek only -- the splice
      must still see the exchange from byte zero.

      This is the one place RTSP can carry a credential without us
      answering anything: proper Basic auth would mean replying 401 and
      renumbering CSeq, which is exactly what makes the opaque splice
      work. ffmpeg passes the query through untouched (verified) and
      its own listener ignores it.

      A password in a URL is normally a bad idea, but this URL is
      configured on an aircraft rather than typed into a browser, so it
      does not end up in history or a Referer header.
     */
    uint8_t line[2048] {};
    std::string pw;
    bool pw_present = false;
    const ssize_t ln = ::recv(fd, line, sizeof(line), MSG_PEEK);
    if (ln <= 0) {
        close(fd);               // fail closed if the peek changed under us
        return;
    }
    const std::string req(reinterpret_cast<char *>(line), size_t(ln));
    const size_t eol = req.find('\n');
    if (eol == std::string::npos) {
        close(fd);               // detect phase promised a complete line
        return;
    }
    std::string first = req.substr(0, eol);
    if (!first.empty() && first.back() == '\r') {
        first.pop_back();
    }
    const size_t sp1 = first.find(' ');
    const size_t sp2 = sp1 == std::string::npos
        ? std::string::npos : first.find(' ', sp1 + 1);
    if (sp1 == std::string::npos || sp2 == std::string::npos) {
        close(fd);
        return;
    }
    const std::string target = first.substr(sp1 + 1, sp2 - sp1 - 1);
    pw_present = http_query_value(target, "pw", pw);

    // The viewer classifier keeps /view/ RTSP playback connections.
    // Everything reaching this splice is still a publisher and must
    // pass publish admission before a backend is started.
    const video_admit_t r = auth_.admit(ke_, uint32_t(from.sin_addr.s_addr),
                                        pw_present ? &pw : nullptr, true,
                                        session_ok(idx), open_publish(idx),
                                        now);
    if (r != VIDEO_ADMIT_OK) {
        log_reject(s, idx, uint32_t(from.sin_addr.s_addr), r, now);
        close(fd);
        return;
    }
    if (s.mkv_fd >= 0 || s.rtsp.running() || s.rtsp_client_fd >= 0
        || (s.has_pub && now - s.pub_last <= VIDEO_PUB_IDLE_S)) {
        log_reject(s, idx, uint32_t(from.sin_addr.s_addr),
                   VIDEO_ADMIT_SLOT_BUSY, now);
        close(fd);
        return;
    }

    const bool want_audio =
        (video_entry_opts(ke_.video_flags) & VIDEO_OPT_AUDIO) != 0;
    if (!s.rtsp.start(port2_, idx, want_audio, proto)) {
        close(fd);
        return;
    }
    fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
    s.rtsp_client_fd = fd;
    s.rtsp_guard.clear();
    s.pub_ip_be = uint32_t(from.sin_addr.s_addr);
    s.pub_port_be = from.sin_port;
    latch_publisher(s, idx, now);
    printf("[%d] video slot %d %s publisher %s\n",
           port2_, idx, splice_proto_name(proto), addr_to_str(from));

    for (int watch : { s.rtsp_client_fd, s.rtsp.backend_fd(),
                       s.rtsp.media_fd() }) {
        if (watch < 0) {
            continue;
        }
        struct epoll_event ev {};
        ev.events = EPOLLIN | EPOLLRDHUP;
        ev.data.fd = watch;
        epoll_ctl(epfd_, EPOLL_CTL_ADD, watch, &ev);
    }
    last_tick_ = 0;
}

void VideoChild::close_rtsp(Slot &s, int idx, const char *why)
{
    // media_fd() covers a backend that reap() has already marked gone:
    // an RTP publisher has no client fd and no RTMP session, so without
    // it the orphaned fd stayed in epoll and spun on EPOLLHUP forever.
    if (!s.rtsp.running() && s.rtsp.media_fd() < 0
        && s.rtsp_client_fd < 0 && !s.rtmp) {
        return;
    }
    // The backend's proto is only meaningful once it started, and an
    // RTMP session can end before that -- during the handshake, or on a
    // refused publish.
    printf("[%d] video slot %d %s publisher gone (%s)\n",
           port2_, idx,
           s.rtmp ? "RTMP" : splice_proto_name(s.rtsp.proto()), why);
    for (int watch : { s.rtsp_client_fd, s.rtsp.backend_fd(),
                       s.rtsp.media_fd() }) {
        if (watch >= 0) {
            epoll_ctl(epfd_, EPOLL_CTL_DEL, watch, nullptr);
        }
    }
    if (s.rtsp_client_fd >= 0) {
        close(s.rtsp_client_fd);
        s.rtsp_client_fd = -1;
    }
    s.rtmp.reset();
    s.to_backend.clear();
    s.to_client.clear();
    s.rtsp_guard.clear();
    s.rtsp.stop();
    s.rec.close_segment();
    s.recording = false;
    s.has_pub = false;
    s.reset_rtp();
    // Same reasoning as the UDP idle-release path: the next publisher
    // is a different stream, so viewers have to be ended rather than
    // spliced onto it. Missing it here left RTSP -- the transport a
    // publish password forces you onto -- with the original bug.
    end_stream(s, idx, "publisher gone");
}

/*
  Move bytes between the publisher and the backend, and pull muxed
  MPEG-TS off the backend's stdout into the normal ingest path.
  Returns false when the session is over.
 */
/*
  Push whatever is queued for one direction. Never waits: a short write
  leaves the remainder queued and EPOLLOUT armed.
 */
bool VideoChild::splice_flush(int to_fd, SpliceQueue &q)
{
    while (q.pending() > 0) {
        const ssize_t w = ::send(to_fd, q.buf.data() + q.sent, q.pending(),
                                 MSG_NOSIGNAL);
        if (w > 0) {
            q.sent += size_t(w);
            continue;
        }
        if (w < 0 && errno == EINTR) {
            continue;
        }
        if (w < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
            break;              // still owed; EPOLLOUT will bring us back
        }
        // A zero return is not progress either -- looping on it spun
        // forever in the previous version, because the offset never
        // advanced.
        return false;
    }
    q.compact();
    return true;
}

/*
  Filter the client-to-backend half of an RTSP splice. The first request
  was already checked before the backend started, but a client can put a
  credential on a later ANNOUNCE. Without continued inspection an absent
  OPTIONS could take session fallback and a later wrong password would be
  silently ignored.
 */
bool VideoChild::guard_rtsp_requests(Slot &s, int idx, const uint8_t *buf,
                                     size_t n, time_t now)
{
    RtspRequestGuard &g = s.rtsp_guard;
    g.buffered.insert(g.buffered.end(), buf, buf + n);

    while (!g.buffered.empty()) {
        if (g.opaque_left > 0) {
            const size_t take = std::min(g.opaque_left, g.buffered.size());
            s.to_backend.buf.insert(s.to_backend.buf.end(),
                                    g.buffered.begin(),
                                    g.buffered.begin() + long(take));
            g.buffered.erase(g.buffered.begin(),
                             g.buffered.begin() + long(take));
            g.opaque_left -= take;
            continue;
        }

        // Interleaved RTP/RTCP: '$', channel, 16-bit big-endian length.
        if (g.buffered[0] == '$') {
            if (g.buffered.size() < 4) {
                return true;
            }
            g.opaque_left = (size_t(g.buffered[2]) << 8) | g.buffered[3];
            s.to_backend.buf.insert(s.to_backend.buf.end(),
                                    g.buffered.begin(),
                                    g.buffered.begin() + 4);
            g.buffered.erase(g.buffered.begin(), g.buffered.begin() + 4);
            continue;
        }

        size_t header_len = 0;
        for (size_t i = 0; i + 1 < g.buffered.size(); i++) {
            if (g.buffered[i] == '\n' && g.buffered[i + 1] == '\n') {
                header_len = i + 2;
                break;
            }
            if (i + 3 < g.buffered.size()
                && g.buffered[i] == '\r' && g.buffered[i + 1] == '\n'
                && g.buffered[i + 2] == '\r'
                && g.buffered[i + 3] == '\n') {
                header_len = i + 4;
                break;
            }
        }
        if (header_len == 0) {
            return g.buffered.size() <= HTTP_MAX_REQUEST;
        }

        HttpRequest req;
        if (req.feed(g.buffered.data(), header_len) != 1) {
            return false;
        }
        std::string pw;
        if (http_query_value(req.target(), "pw", pw)) {
            bool have_pw = false;
            for (uint8_t b : ke_.video_publish_key) {
                have_pw |= b != 0;
            }
            bool matches = video_password_matches(ke_.video_publish_key, pw);
            /*
              ffmpeg resolves an SDP control path after the whole source
              URI, producing e.g. ?pw=secret/streamid=0. URI syntax makes
              that suffix part of the query value. Accept it only when a
              slash-delimited prefix is itself the exact password; the
              peer still has to know the configured credential.
             */
            const size_t slash = pw.rfind('/');
            if (!matches && slash != std::string::npos) {
                matches = video_password_matches(
                    ke_.video_publish_key, pw.substr(0, slash));
            }
            if (have_pw && !matches) {
                log_reject(s, idx, s.pub_ip_be, VIDEO_ADMIT_BAD_PASSWORD,
                           now);
                return false;
            }
        }

        const size_t content_length_count =
            req.header_count("Content-Length");
        const std::string content_length = req.header("Content-Length");
        if (content_length_count > 1
            || (content_length_count == 1 && content_length.empty())) {
            return false;
        }
        if (!content_length.empty()) {
            char *end = nullptr;
            errno = 0;
            const unsigned long long body =
                strtoull(content_length.c_str(), &end, 10);
            if (errno != 0 || end == content_length.c_str() || *end != '\0'
                || body > 16u * 1024u * 1024u) {
                return false;
            }
            g.opaque_left = size_t(body);
        }
        s.to_backend.buf.insert(s.to_backend.buf.end(),
                                g.buffered.begin(),
                                g.buffered.begin() + long(header_len));
        g.buffered.erase(g.buffered.begin(),
                         g.buffered.begin() + long(header_len));
    }
    return true;
}

// EPOLLOUT only while something is queued. Armed unconditionally it
// would make epoll_wait return immediately for ever on an idle splice,
// which is the same idle-viewer CPU burn measured earlier.
void VideoChild::splice_arm(int to_fd, SpliceQueue &q)
{
    const bool want = q.pending() > 0;
    if (want == q.armed || to_fd < 0) {
        return;
    }
    struct epoll_event ev {};
    ev.events = uint32_t(EPOLLIN | EPOLLRDHUP)
        | (want ? uint32_t(EPOLLOUT) : 0u);
    ev.data.fd = to_fd;
    if (epoll_ctl(epfd_, EPOLL_CTL_MOD, to_fd, &ev) == 0) {
        q.armed = want;
    }
}

/*
  Close one pending handshake and forget it. It owns no slot state, so
  there is nothing else to unwind.
 */
void VideoChild::close_pending(Slot &s, int idx, PendingRtmp &p,
                               const char *why)
{
    (void)s;
    if (!p.active()) {
        return;
    }
    struct in_addr a {};
    a.s_addr = p.ip_be;
    printf("[%d] video slot %d RTMP handshake from %s ended (%s)\n",
           port2_, idx, inet_ntoa(a), why);
    epoll_ctl(epfd_, EPOLL_CTL_DEL, p.fd, nullptr);
    close(p.fd);
    p.fd = -1;
    p.sess.reset();
    p.out.clear();
    p.since = 0;
}

/*
  A pending handshake has published. Authorise it and, if the slot is
  free, hand it over.

  Returns false if this connection is finished either way.
 */
bool VideoChild::promote_pending(Slot &s, int idx, PendingRtmp &p,
                                 time_t now)
{
    RtmpSession &r = *p.sess;

    const video_admit_t a = auth_.admit(
        ke_, p.ip_be, r.password_present() ? &r.password() : nullptr, true,
        session_ok(idx), open_publish(idx), now);
    if (a != VIDEO_ADMIT_OK) {
        log_reject(s, idx, p.ip_be, a, now);
        r.reject_publish("NetStream.Publish.Denied", video_admit_str(a));
        return false;
    }

    /*
      An RTMP path on the slot is an optional restriction: we read the
      app and stream off the wire rather than telling ffmpeg what to
      expect. Left blank the slot takes whatever is published.
     */
    char want[sizeof(ke_.video_rtmp_path[0]) + 1] {};
    memcpy(want, video_rtmp_path_of(ke_, unsigned(idx)),
           sizeof(ke_.video_rtmp_path[0]));
    want[sizeof(want) - 1] = '\0';
    if (want[0] != '\0' && r.path() != want) {
        printf("[%d] video slot %d RTMP publisher refused: published %s, "
               "slot expects %s\n", port2_, idx, r.path().c_str(), want);
        r.reject_publish("NetStream.Publish.Denied",
                         "stream path does not match this slot");
        return false;
    }

    // Authorised -- but the slot may have been taken while this one was
    // still negotiating.
    if (s.mkv_fd >= 0 || s.rtsp.running() || s.rtsp_client_fd >= 0
        || (s.has_pub && now - s.pub_last <= VIDEO_PUB_IDLE_S)) {
        log_reject(s, idx, p.ip_be, VIDEO_ADMIT_SLOT_BUSY, now);
        r.reject_publish("NetStream.Publish.Denied",
                         "another publisher holds this slot");
        return false;
    }

    // Hand the socket and the session to the slot.
    s.rtsp_client_fd = p.fd;
    s.rtmp = std::move(p.sess);
    s.to_client = std::move(p.out);
    s.to_backend.clear();
    p.fd = -1;
    p.out.clear();
    p.since = 0;

    s.pub_ip_be = p.ip_be;
    s.pub_port_be = p.port_be;
    s.rtmp->accept_publish();
    latch_publisher(s, idx, now);
    printf("[%d] video slot %d RTMP publishing %s\n",
           port2_, idx, s.rtmp->path().c_str());
    /*
      Parsing stopped at publish so this authorisation could happen in
      order. Whatever the publisher pipelined behind it -- for a client
      that does not wait for onStatus, that includes the sequence header
      and its parameter sets -- is still buffered, so pick it up now
      rather than waiting for a read that may never come.
     */
    if (!rtmp_drain_owner(s, idx, s.rtmp->resume())) {
        close_rtsp(s, idx, "RTMP session ended during promotion");
        return true;
    }
    /*
      Push the responses out. accept_publish() queued the
      NetStream.Publish.Start the client is waiting on, and nothing else
      runs until the next epoll event -- which a publisher that waits
      for onStatus before sending anything will never cause. Leaving it
      queued is exactly the stall this whole path exists to fix.
     */
    if (!splice_flush(s.rtsp_client_fd, s.to_client)) {
        close_rtsp(s, idx, "connection closed");
        return true;
    }
    splice_arm(s.rtsp_client_fd, s.to_client);
    return true;
}

/*
  Drive one pending handshake. Returns false when it is finished.
 */
bool VideoChild::pump_pending(Slot &s, int idx, PendingRtmp &p, time_t now)
{
    if (!splice_flush(p.fd, p.out)) {
        return false;
    }
    uint8_t buf[8192];
    const ssize_t n = ::recv(p.fd, buf, sizeof(buf), 0);
    if (n == 0) {
        return false;
    }
    if (n < 0) {
        if (!(errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) {
            return false;
        }
        splice_arm(p.fd, p.out);
        return true;
    }

    RtmpSession &r = *p.sess;
    bool alive = r.feed(buf, size_t(n), now);
    bool promoted = false;
    if (r.publish_pending()) {
        promoted = promote_pending(s, idx, p, now);
        if (promoted) {
            return false;   // the slot owns it now; stop pumping as pending
        }
        alive = false;      // refused: flush the status, then close
    }
    if (!r.to_peer().empty()) {
        p.out.buf.insert(p.out.buf.end(), r.to_peer().begin(),
                         r.to_peer().end());
        r.to_peer().clear();
    }
    // Media before publish is not ours to keep, and the session drops it.
    r.to_flv().clear();
    if (!splice_flush(p.fd, p.out)) {
        return false;
    }
    if (!alive) {
        if (r.error()[0] != '\0') {
            printf("[%d] video slot %d RTMP: %s\n", port2_, idx, r.error());
        }
        return false;
    }
    splice_arm(p.fd, p.out);
    return true;
}

/*
  Start the backend now that the publisher's codec is known. Returns
  false if it could not be started.
 */
bool VideoChild::rtmp_start_backend(Slot &s, int idx)
{
    /*
      h264_metadata rewrites the NAL units, which drops the zero-length
      one ffmpeg's own AVCC to Annex-B conversion puts ahead of every
      access unit for this camera. Chrome's MP4 parser refuses a sample
      containing it ("Failed to prepare video sample for decode");
      Firefox plays it. It is H.264-only, hence the codec test.
     */
    const char *vbsf = s.rtmp->video_codec() == RTMP_VCODEC_H264
        ? "h264_metadata" : nullptr;
    const bool want_audio =
        (video_entry_opts(ke_.video_flags) & VIDEO_OPT_AUDIO) != 0;
    if (!s.rtsp.start(port2_, idx, want_audio, SPLICE_RTMP, vbsf)) {
        return false;
    }
    for (int watch : { s.rtsp.backend_fd(), s.rtsp.media_fd() }) {
        if (watch < 0) {
            continue;
        }
        struct epoll_event ev {};
        ev.events = EPOLLIN | EPOLLRDHUP;
        ev.data.fd = watch;
        epoll_ctl(epfd_, EPOLL_CTL_ADD, watch, &ev);
    }
    return true;
}

/*
  Drive a native RTMP publisher: client bytes in, responses out, FLV to
  the backend. Returns false when the session is over.
 */
/*
  Move what the owning session produced into the queues, and start the
  backend once the stream has named its codec.

  `alive` is what feed()/resume() returned. Shared with promotion,
  which has to run exactly this after accepting a publish: the client
  may have pipelined its sequence header into the same segment, and
  parsing stops at publish so authorisation can happen in order.
 */
bool VideoChild::rtmp_drain_owner(Slot &s, int idx, bool alive)
{
    RtmpSession &r = *s.rtmp;
    // A second publish on the owning connection is a protocol error the
    // session rejects; there is nothing to authorise here, because
    // promotion did that before this session reached the slot.
    const bool refused = r.publish_pending();

    if (!r.to_peer().empty()) {
        s.to_client.buf.insert(s.to_client.buf.end(),
                               r.to_peer().begin(), r.to_peer().end());
        r.to_peer().clear();
    }
    /*
      Queue unconditionally. The backend may not exist yet -- it waits
      for the codec -- and the FLV header and sequence header arrive
      before it does; dropping them left the backend with a stream it
      could not open. Reads are gated on SPLICE_QUEUE_MAX, so this stays
      bounded.
     */
    if (!r.to_flv().empty()) {
        s.to_backend.buf.insert(s.to_backend.buf.end(),
                                r.to_flv().begin(), r.to_flv().end());
        r.to_flv().clear();
    }
    if (r.publishing() && !s.rtsp.running()
        && r.video_codec() != RTMP_VCODEC_NONE
        && !rtmp_start_backend(s, idx)) {
        return false;
    }
    if (refused || !alive) {
        splice_flush(s.rtsp_client_fd, s.to_client);
        if (r.error()[0] != '\0') {
            printf("[%d] video slot %d RTMP: %s\n", port2_, idx, r.error());
        }
        return false;
    }
    return true;
}

bool VideoChild::pump_rtmp(Slot &s, int idx, int fd, time_t now)
{
    uint8_t buf[16384];
    if (fd == s.rtsp.media_fd()) {
        const ssize_t n = ::read(fd, buf, sizeof(buf));
        if (n == 0) {
            return false;
        }
        if (n < 0) {
            return errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR;
        }
        s.pub_last = now;
        ingest_stream(s, idx, buf, size_t(n));
        return true;
    }

    // The backend's stdin: write-only, so an event here is either room
    // to write or the backend having gone away.
    if (fd == s.rtsp.backend_fd()) {
        if (!splice_flush(fd, s.to_backend)) {
            return false;
        }
        splice_arm(fd, s.to_backend);
        return true;
    }

    if (fd != s.rtsp_client_fd || !s.rtmp) {
        return true;
    }

    if (!splice_flush(fd, s.to_client)) {
        return false;
    }
    if (s.rtsp.running() && !splice_flush(s.rtsp.backend_fd(), s.to_backend)) {
        return false;
    }

    /*
      Read only while both queues have room. Gating on the backend alone
      let a peer that stops reading its responses keep sending commands,
      turning its own bounded input into an unbounded to_client.
     */
    if (s.to_backend.pending() < SPLICE_QUEUE_MAX
        && s.to_client.pending() < SPLICE_QUEUE_MAX) {
        const ssize_t n = ::recv(fd, buf, sizeof(buf), 0);
        if (n == 0) {
            return false;
        }
        if (n < 0 && !(errno == EAGAIN || errno == EWOULDBLOCK
                       || errno == EINTR)) {
            return false;
        }
        if (n > 0) {
            s.pub_last = now;
            if (!rtmp_drain_owner(s, idx, s.rtmp->feed(buf, size_t(n), now))) {
                return false;
            }
        }
    }

    if (!splice_flush(fd, s.to_client)) {
        return false;
    }
    if (s.rtsp.running()) {
        if (!splice_flush(s.rtsp.backend_fd(), s.to_backend)) {
            return false;
        }
        splice_arm(s.rtsp.backend_fd(), s.to_backend);
    }
    splice_arm(fd, s.to_client);
    return true;
}

bool VideoChild::pump_rtsp(Slot &s, int idx, int fd, time_t now)
{
    if (s.rtmp) {
        return pump_rtmp(s, idx, fd, now);
    }
    uint8_t buf[16384];
    if (fd == s.rtsp_client_fd || fd == s.rtsp.backend_fd()) {
        const int from_fd = fd;
        const int to_fd = (fd == s.rtsp_client_fd) ? s.rtsp.backend_fd()
                                                   : s.rtsp_client_fd;
        if (to_fd < 0) {
            return false;
        }
        SpliceQueue &out = (fd == s.rtsp_client_fd) ? s.to_backend
                                                    : s.to_client;
        SpliceQueue &in = (fd == s.rtsp_client_fd) ? s.to_client
                                                   : s.to_backend;

        // Drain anything owed in both directions first: this fd may have
        // woken us for EPOLLOUT rather than EPOLLIN.
        if (!splice_flush(to_fd, out)) {
            return false;
        }
        if (!splice_flush(from_fd, in)) {
            return false;
        }

        // Only read while the sink has room. Leaving bytes in the socket
        // is what pushes back on the publisher, instead of buffering
        // without limit here.
        if (out.pending() < SPLICE_QUEUE_MAX) {
            const ssize_t n = ::recv(from_fd, buf, sizeof(buf), 0);
            if (n == 0) {
                return false;
            }
            if (n < 0 && !(errno == EAGAIN || errno == EWOULDBLOCK
                           || errno == EINTR)) {
                return false;
            }
            if (n > 0) {
                if (fd == s.rtsp_client_fd) {
                    if (!guard_rtsp_requests(s, idx, buf, size_t(n), now)) {
                        return false;
                    }
                } else {
                    out.buf.insert(out.buf.end(), buf, buf + n);
                }
                if (!splice_flush(to_fd, out)) {
                    return false;
                }
                s.pub_last = now;
            }
        }

        splice_arm(to_fd, out);
        splice_arm(from_fd, in);
        return true;
    }
    if (fd == s.rtsp.media_fd()) {
        const ssize_t n = ::read(fd, buf, sizeof(buf));
        if (n == 0) {
            return false;
        }
        if (n < 0) {
            return errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR;
        }
        s.pub_last = now;
        // The backend writes a byte stream, not datagrams, so packets
        // straddle reads -- which the scanner and ring already handle.
        ingest_stream(s, idx, buf, size_t(n));
        return true;
    }
    return true;
}

void VideoChild::epoll_add_viewer(VideoViewer &v, uint32_t &events)
{
    struct epoll_event ev {};
    ev.events = EPOLLIN | EPOLLRDHUP;
    ev.data.fd = v.fd();
    if (epoll_ctl(epfd_, EPOLL_CTL_ADD, v.fd(), &ev) == 0) {
        events = ev.events;
    }
}

/*
  Keep the epoll mask matched to what the viewer needs. EPOLLOUT only
  while it is blocked: armed permanently it makes epoll_wait return
  immediately for any writable socket, a core per idle viewer. EPOLLET
  only while it is holding unconsumed bytes waiting for more: level
  triggered, those bytes re-fire EPOLLIN on every wait, a core per
  split request line until the detect timeout.
 */
void VideoChild::epoll_sync_viewer(VideoViewer &v, uint32_t &events)
{
    const uint32_t want = uint32_t(EPOLLIN | EPOLLRDHUP)
        | (v.wants_write() ? uint32_t(EPOLLOUT) : 0u)
        | (v.holding_bytes() ? uint32_t(EPOLLET) : 0u);
    if (want == events) {
        return;
    }
    struct epoll_event ev {};
    ev.events = want;
    ev.data.fd = v.fd();
    if (epoll_ctl(epfd_, EPOLL_CTL_MOD, v.fd(), &ev) == 0) {
        events = want;
    }
}

void VideoChild::drop_viewer(Slot &s, int idx, VideoViewer &v)
{
    if (!v.active()) {
        return;
    }
    printf("[%d] video slot %d viewer disconnected after %llu KiB (%s)\n",
           port2_, idx, (unsigned long long)(v.bytes_sent() / 1024),
           v.drop_reason()[0] ? v.drop_reason() : "closed");
    epoll_ctl(epfd_, EPOLL_CTL_DEL, v.fd(), nullptr);
    v.close();
    s.viewers_dropped++;
    last_tick_ = 0;
}

/*
  End the current stream on a slot.

  A publisher going away ends the stream its viewers are watching. The
  next publisher is a *different* stream: PSI, continuity counters and
  PTS/PCR all restart, so its bytes cannot simply be appended to what
  the viewers have already been given. A player fed that sees time jump
  backwards and stalls for good -- which looks exactly like "the video
  never came back" rather than like a disconnect.

  So viewers are closed here and get a clean end of stream. A client
  that wants to keep watching reconnects and joins the new stream at its
  own first keyframe. The ring and scanner are reset with them, so no
  stale anchor from the old stream can be handed to whoever joins next.
 */
void VideoChild::end_stream(Slot &s, int idx, const char *why)
{
    for (int v = 0; v < VIDEO_MAX_VIEWERS; v++) {
        VideoViewer &vw = s.viewers[v];
        if (vw.active()) {
            vw.set_drop_reason(why);
            drop_viewer(s, idx, vw);
        }
    }
    s.ring.reset();
    s.scanner.reset();
    s.last_anchor = 0;
    s.had_anchor = false;
    s.bad_datagrams = 0;
    s.warned_204 = false;
    s.pub_bytes = 0;
}

// Push ring bytes to every viewer on this slot. Called after ingest and
// on every loop iteration, so a viewer whose socket drained between
// epoll wakeups still makes progress.
void VideoChild::pump_viewers(Slot &s, int idx, time_t now)
{
    for (int v = 0; v < VIDEO_MAX_VIEWERS; v++) {
        VideoViewer &vw = s.viewers[v];
        if (!vw.active()) {
            continue;
        }
        if (vw.kind() == VVK_RTSP || vw.kind() == VVK_RTMP || vw.kind() == VVK_MKV) {
            // A publisher, not a viewer. Take the socket -- untouched,
            // since the detect phase only ever peeked -- and splice it.
            struct sockaddr_in from {};
            from.sin_family = AF_INET;
            from.sin_addr.s_addr = vw.peer_ip_be();
            from.sin_port = vw.peer_port_be();
            const splice_proto_t proto = vw.kind() == VVK_RTMP
                ? SPLICE_RTMP : SPLICE_RTSP;
            const bool mkv = vw.kind() == VVK_MKV;
            const int fd = vw.release_fd();
            epoll_ctl(epfd_, EPOLL_CTL_DEL, fd, nullptr);
            s.viewer_events[v] = 0;
            if (mkv) handle_mkv(s, idx, fd, from, now);
            else handle_rtsp(s, idx, fd, from, now, proto);
            continue;
        }
        if (vw.state() == VV_DETECT && vw.kind() == VVK_WS) {
            if (!vw.begin_ws_pump(ke_, idx, s.ring, s.scanner, now)) {
                drop_viewer(s, idx, vw);
            }
            continue;
        }
        if (vw.state() == VV_DETECT) {
            if (now - vw.connected_at() >= VIDEO_DETECT_SILENCE_S) {
                if (!vw.detect_timeout(ke_, idx, s.ring, s.scanner, now)) {
                    drop_viewer(s, idx, vw);
                    continue;
                }
            } else {
                continue;
            }
        }
        if (!vw.on_writable(s.ring, now)) {
            drop_viewer(s, idx, vw);
            continue;
        }
        epoll_sync_viewer(vw, s.viewer_events[v]);
    }
}

void VideoChild::write_conn_rows(time_t now)
{
    auto *db = conn_db_open_transaction();
    if (db == nullptr) {
        return;
    }
    // Only our own index range: the MAVLink child snapshots 0..999 the
    // same way, and whole-port2 deletes here would erase its rows.
    conn_delete_index_range(db, port2_, VIDEO_CONN_INDEX_BASE, INT32_MAX);
    for (int i = 0; i < KEY_MAX_VIDEO_PORTS; i++) {
        Slot &s = slots_[i];
        if (!s.has_pub) {
            continue;
        }
        struct ConnEntry e {};
        e.magic = CONN_MAGIC;
        e.connected_at = uint64_t(s.pub_since);
        e.last_update = uint64_t(now);
        e.port2 = port2_;
        e.conn_index = VIDEO_PUB_INDEX(i);
        e.pid = uint32_t(getpid());
        e.rx_msgs = uint32_t(s.pub_bytes / 1024);   // KiB for video rows
        e.peer_ip_be = s.pub_ip_be;
        e.peer_port_be = s.pub_port_be;
        /*
          Report what the publisher actually is. Every video row used to
          say UDP/MPEG-TS, which is only true of the datagram path -- an
          operator looking at a stuck RTSP or RTMP publisher was told it
          was something it was not.
         */
        const bool rtp_pub = s.rtsp.running()
                             && s.rtsp.proto() == SPLICE_RTP;
        const bool tcp_pub = !rtp_pub && (s.mkv_fd >= 0 || s.rtmp || s.rtsp.running()
                                          || s.rtsp_client_fd >= 0);
        e.transport = tcp_pub ? CONN_TRANSPORT_TCP : CONN_TRANSPORT_UDP;
        e.is_user = 0;
        e.role = CONN_ROLE_VIDEO_PUB;
        e.stream_idx = uint8_t(i);
        e.app_proto = s.mkv_fd >= 0 ? CONN_APP_MATROSKA : s.rtmp ? CONN_APP_RTMP
                    : rtp_pub ? CONN_APP_RTP
                    : tcp_pub ? CONN_APP_RTSP : CONN_APP_MPEGTS;
        conn_write(db, e);
    }
    conn_db_close_commit(db);
}

void VideoChild::tick(time_t now)
{
    // Policy that doesn't need a rebind (credentials, grace, quota) is
    // picked up here. Anything that changes binding -- the enable bit,
    // the ports, the slot options -- makes the parent re-fork us
    // instead, so we never have to rebind under our own feet.
    if (load_entry()) {
        auth_.invalidate();
    }
    // Sample the MAVLink session while it is alive, so the grace window
    // still has a last-known-good to age out once the session child
    // exits and its connections.tdb row disappears.
    auth_.observe(now);
    for (int i = 0; i < KEY_MAX_VIDEO_PORTS; i++) {
        Slot &s = slots_[i];
        if (s.mkv_fd >= 0 && now-s.pub_last > VIDEO_PUB_IDLE_S) close_mkv(s, i, "publisher idle");
        if (s.rtsp.running() && s.rtsp.reap()) {
            close_rtsp(s, i, "backend exited");
        }
        /*
          Time out handshakes that never publish. The session bounds
          itself once bytes arrive; this covers the peer that connects
          and then says nothing at all.
         */
        for (int k = 0; k < VIDEO_MAX_PENDING_RTMP; k++) {
            PendingRtmp &p = s.pending[k];
            if (p.active() && p.since != 0
                && now - p.since > RTMP_PREPUBLISH_MAX_S) {
                close_pending(s, i, p, "handshake timed out");
            }
        }
        /*
          Accepted, but never said what it is sending. There is no
          backend and no media, yet its control messages keep refreshing
          the idle timer, so nothing else would ever reclaim the slot.
         */
        if (s.rtmp && s.rtmp->publishing() && !s.rtsp.running()
            && s.rtmp->publishing_since() != 0
            && now - s.rtmp->publishing_since() > RTMP_CODEC_DEADLINE_S) {
            close_rtsp(s, i, "no video codec after publish");
            continue;
        }
        if (s.has_pub && now - s.pub_last > VIDEO_PUB_IDLE_S) {
            printf("[%d] video slot %d publisher idle, releasing\n",
                   port2_, i);
            /*
              A spliced publisher has to be torn down, not just
              forgotten. Clearing has_pub alone left the client socket,
              the backend socket and the ffmpeg all alive, so
              rtsp.running() went on refusing every replacement as
              slot-busy until the child was killed by hand -- and UDP
              admission, which tests only has_pub, could meanwhile claim
              the slot underneath a splice that was still feeding it.
             */
            if (s.rtsp.running() || s.rtsp_client_fd >= 0) {
                close_rtsp(s, i, "publisher idle");
                continue;    // close_rtsp does the rest, including
                             // end_stream and the recorder
            }
            s.has_pub = false;
            s.reset_rtp();
            // Close the segment on disconnect rather than leaving it
            // open: the file is complete, and an open file is not
            // evictable by the quota pass.
            s.rec.close_segment();
            s.recording = false;
            end_stream(s, i, "publisher gone");
        }
    }
    // One line per active publisher. This is the only window onto the
    // scanner until viewers exist, so it carries what a operator (and
    // the tests) need: whether a viewer could join, and why not.
    for (int i = 0; i < KEY_MAX_VIDEO_PORTS; i++) {
        Slot &s = slots_[i];
        if (!s.has_pub) {
            continue;
        }
        const TSStats &st = s.scanner.stats();
        uint64_t join = 0;
        const bool can_join = s.scanner.join_offset(join);
        printf("[%d] video slot %d stats: %llu KiB, %llu pkts, pat=%llu "
               "pmt=%llu rai=%llu cc_err=%llu crc_err=%llu bad_dgram=%llu "
               "vpid=0x%x stype=0x%02x join=%s\n",
               port2_, i,
               (unsigned long long)(s.pub_bytes / 1024),
               (unsigned long long)st.packets,
               (unsigned long long)st.pat_seen,
               (unsigned long long)st.pmt_seen,
               (unsigned long long)st.rai_seen,
               (unsigned long long)st.cc_errors,
               (unsigned long long)st.crc_errors,
               (unsigned long long)s.bad_datagrams,
               unsigned(s.scanner.video_pid()),
               unsigned(s.scanner.video_stream_type()),
               can_join ? "ready" : "waiting");
    }
    write_conn_rows(now);
    last_tick_ = now;
}

void VideoChild::run(void)
{
    if (!load_entry()) {
        printf("[%d] video: no keys.tdb entry\n", port2_);
        signal_ready(ENOENT);
        _exit(1);
    }

    int err = bind_slots();
    signal_ready(err);

    epfd_ = epoll_create1(0);
    if (epfd_ == -1) {
        printf("[%d] video: epoll_create1 failed - %s\n",
               port2_, strerror(errno));
        _exit(1);
    }
    for (int i = 0; i < KEY_MAX_VIDEO_PORTS; i++) {
        for (int fd : { slots_[i].udp_fd, slots_[i].tcp_fd }) {
            if (fd == -1) {
                continue;
            }
            struct epoll_event ev {};
            ev.events = EPOLLIN;
            ev.data.fd = fd;
            epoll_ctl(epfd_, EPOLL_CTL_ADD, fd, &ev);
        }
    }

    while (true) {
        struct epoll_event events[VIDEO_MAX_EPOLL_EVENTS];
        int nev = epoll_wait(epfd_, events, VIDEO_MAX_EPOLL_EVENTS, 1000);
        if (nev == -1 && errno != EINTR) {
            printf("[%d] video: epoll_wait failed - %s\n",
                   port2_, strerror(errno));
            break;
        }
        const time_t evnow = time(nullptr);
        for (int e = 0; e < nev; e++) {
            const int fd = events[e].data.fd;
            bool matched = false;
            for (int i = 0; i < KEY_MAX_VIDEO_PORTS && !matched; i++) {
                if (slots_[i].udp_fd == fd) {
                    handle_udp(slots_[i], i);
                    matched = true;
                } else if (slots_[i].tcp_fd == fd) {
                    handle_tcp(slots_[i], i);
                    matched = true;
                }
            }
            if (matched) {
                continue;
            }
            // a pending RTMP handshake: owns no slot state, so it is
            // matched before the splice fds and closed on its own.
            for (int i = 0; i < KEY_MAX_VIDEO_PORTS && !matched; i++) {
                Slot &s = slots_[i];
                for (int k = 0; k < VIDEO_MAX_PENDING_RTMP; k++) {
                    PendingRtmp &p = s.pending[k];
                    if (!p.active() || p.fd != fd) {
                        continue;
                    }
                    matched = true;
                    if ((events[e].events & (EPOLLHUP | EPOLLERR))
                        || !pump_pending(s, i, p, evnow)) {
                        // Promotion moves the fd out and clears p, so an
                        // entry that is no longer active was handed on
                        // rather than dropped.
                        if (p.active()) {
                            close_pending(s, i, p, "connection closed");
                        }
                    }
                    break;
                }
            }
            if (matched) {
                continue;
            }
            for (int i=0; i<KEY_MAX_VIDEO_PORTS && !matched; i++) {
                Slot &s = slots_[i];
                if (s.mkv_fd != fd) continue;
                matched = true;
                if ((events[e].events & EPOLLERR) || !pump_mkv(s, i, evnow))
                    close_mkv(s, i, "Matroska publisher disconnected or malformed");
            }
            if (matched) continue;
            // an RTSP splice fd
            for (int i = 0; i < KEY_MAX_VIDEO_PORTS && !matched; i++) {
                Slot &s = slots_[i];
                if (fd != s.rtsp_client_fd && fd != s.rtsp.backend_fd()
                    && fd != s.rtsp.media_fd()) {
                    continue;
                }
                matched = true;
                if ((events[e].events & (EPOLLHUP | EPOLLERR))
                    || !pump_rtsp(s, i, fd, evnow)) {
                    close_rtsp(s, i, "connection closed");
                }
            }
            if (matched) {
                continue;
            }
            // a viewer socket
            for (int i = 0; i < KEY_MAX_VIDEO_PORTS && !matched; i++) {
                Slot &s = slots_[i];
                for (int v = 0; v < VIDEO_MAX_VIEWERS; v++) {
                    VideoViewer &vw = s.viewers[v];
                    if (!vw.active() || vw.fd() != fd) {
                        continue;
                    }
                    matched = true;
                    bool ok = true;
                    if (events[e].events & (EPOLLHUP | EPOLLERR)) {
                        ok = false;
                    } else {
                        if (events[e].events & (EPOLLIN | EPOLLRDHUP)) {
                            ok = vw.on_readable(ke_, i, s.ring, s.scanner,
                                                evnow);
                        }
                        if (ok && (events[e].events & EPOLLOUT)) {
                            ok = vw.on_writable(s.ring, evnow);
                        }
                    }
                    if (!ok) {
                        drop_viewer(s, i, vw);
                    } else {
                        // The push loop below skips connections still
                        // in detect, so this is where a held partial
                        // line switches the fd to edge-triggered.
                        epoll_sync_viewer(vw, s.viewer_events[v]);
                    }
                    break;
                }
            }
        }

        // Push to viewers every iteration, not only on EPOLLOUT: new
        // ring data has no fd event of its own, and a viewer whose
        // socket had room all along would otherwise never be fed.
        const time_t now = time(nullptr);
        for (int i = 0; i < KEY_MAX_VIDEO_PORTS; i++) {
            if (slots_[i].udp_fd != -1 || slots_[i].tcp_fd != -1) {
                pump_viewers(slots_[i], i, now);
            }
        }
        if (now - last_tick_ >= VIDEO_TICK_S) {
            tick(now);
        }
        // The parent may have died between our PDEATHSIG being armed
        // and now; PDEATHSIG only fires on a parent that was alive when
        // it was set.
        if (getppid() == 1) {
            break;
        }
    }

    conn_remove_video(port2_);
    _exit(0);
}

}  // namespace

void video_child_main(int port2, int ready_fd)
{
    VideoChild child(port2, ready_fd);
    child.run();
}