⭐ feat(runner): Fixes and improvements (#259)

## Description

- Improves latency for runner
- Fixes bugs in entrypoint bash scripts
- Package updates, gstreamer 1.26 and workaround for it

Modified runner workflow to hopefully pull latest cachyos base image on
nightlies. This will cause a full build but for nightlies should be
fine?

Also removed the duplicate key-down workaround as we've enabled ordered
datachannels now. Increased retransmit to 2 from 0 to see if it'll help
with some network issues.

Marked as draft as I need to do bug testing still, I'll do it after
fever calms down 😅



<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

- **New Features**
- Enhanced deployment workflows with optimized container image
management.
- Improved audio and video processing for lower latency and better
synchronization.
  - Consolidated debugging options to ease command-line monitoring.

- **Refactor**
- Streamlined internal script flow and process handling for smoother
performance.
- Updated dependency management and communication protocols to boost
overall stability.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Co-authored-by: DatCaptainHorse <DatCaptainHorse@users.noreply.github.com>

.github/workflows/runner.yml

@@ -27,6 +27,7 @@ env:
     REGISTRY: ghcr.io
     IMAGE_NAME: nestrilabs/nestri
     BASE_TAG_PREFIX: runner
+    BASE_IMAGE: docker.io/cachyos/cachyos:latest
 
 # This makes our release ci quit prematurely
 # concurrency:
@@ -55,7 +56,7 @@ jobs:
                 swap-size-gb: 20
             - 
               name: Build Docker image
-              uses: docker/build-push-action@v5
+              uses: docker/build-push-action@v6
               with:
                 file: containers/runner.Containerfile
                 context: ./
@@ -107,7 +108,7 @@ jobs:
                 swap-size-gb: 20
             - 
               name: Build Docker image
-              uses: docker/build-push-action@v5
+              uses: docker/build-push-action@v6
               with:
                 file:  containers/runner.Containerfile
                 context: ./
@@ -116,3 +117,4 @@ jobs:
                 labels: ${{ steps.meta.outputs.labels }}
                 cache-from: type=gha,mode=max
                 cache-to: type=gha,mode=max
+                pull: ${{ github.event_name == 'schedule' }} # Pull base image for scheduled builds

containers/runner.Containerfile

@@ -85,8 +85,8 @@ RUN --mount=type=cache,target=/var/cache/pacman/pkg \
     pacman -Sy --noconfirm meson pkgconf cmake git gcc make \
     libxkbcommon wayland gstreamer gst-plugins-base gst-plugins-good libinput
 
-# Clone repository with proper directory structure
-RUN git clone -b dev-dmabuf https://github.com/games-on-whales/gst-wayland-display.git
+# Clone repository
+RUN git clone -b dev-dmabuf https://github.com/DatCaptainHorse/gst-wayland-display.git
 
 #--------------------------------------------------------------------
 FROM gst-wayland-deps AS gst-wayland-planner
@@ -133,8 +133,8 @@ RUN sed -i \
 RUN --mount=type=cache,target=/var/cache/pacman/pkg \
     pacman -Sy --needed --noconfirm \
         vulkan-intel lib32-vulkan-intel vpl-gpu-rt mesa \
-        steam steam-native-runtime \
-        sudo xorg-xwayland seatd libinput labwc wlr-randr mangohud \
+        steam steam-native-runtime gtk3 lib32-gtk3 \
+        sudo xorg-xwayland seatd libinput labwc wlr-randr gamescope mangohud \
         libssh2 curl wget \
         pipewire pipewire-pulse pipewire-alsa wireplumber \
         noto-fonts-cjk supervisor jq chwd lshw pacman-contrib && \
@@ -144,6 +144,9 @@ RUN --mount=type=cache,target=/var/cache/pacman/pkg \
         gst-plugins-bad gst-plugin-pipewire \
         gst-plugin-webrtchttp gst-plugin-rswebrtc gst-plugin-rsrtp \
         gst-plugin-va gst-plugin-qsv && \
+    # lib32 GStreamer stack to fix some games with videos
+    pacman -Sy --needed --noconfirm \
+        lib32-gstreamer lib32-gst-plugins-base lib32-gst-plugins-good && \
     # Cleanup
     paccache -rk1 && \
     rm -rf /usr/share/{info,man,doc}/*
@@ -185,6 +188,30 @@ RUN mkdir -p /run/dbus && \
         -e '/wants = \[/{s/hooks\.node\.suspend\s*//; s/,\s*\]/]/}' \
         /usr/share/wireplumber/wireplumber.conf
 
+### PipeWire Latency Optimizations (1-5ms instead of 20ms) ###
+RUN mkdir -p /etc/pipewire/pipewire.conf.d && \
+    echo "[audio]\
+    \n  default.clock.rate = 48000\
+    \n  default.clock.quantum = 128\
+    \n  default.clock.min-quantum = 128\
+    \n  default.clock.max-quantum = 256" > /etc/pipewire/pipewire.conf.d/low-latency.conf && \
+    mkdir -p /etc/wireplumber/main.lua.d && \
+    echo 'table.insert(default_nodes.rules, {\
+    \n  matches = { { { "node.name", "matches", ".*" } } },\
+    \n  apply_properties = {\
+    \n    ["audio.format"] = "S16LE",\
+    \n    ["audio.rate"] = 48000,\
+    \n    ["audio.channels"] = 2,\
+    \n    ["api.alsa.period-size"] = 128,\
+    \n    ["api.alsa.headroom"] = 0,\
+    \n    ["session.suspend-timeout-seconds"] = 0\
+    \n  }\
+    \n})' > /etc/wireplumber/main.lua.d/50-low-latency.lua && \
+    echo "default-fragments = 2\
+    \ndefault-fragment-size-msec = 2" >> /etc/pulse/daemon.conf && \
+    echo "load-module module-loopback latency_msec=1" >> /etc/pipewire/pipewire.conf.d/loopback.conf
+
+
 ### Artifacts and Verification ###
 COPY --from=nestri-server-cached-builder /artifacts/nestri-server /usr/bin/
 COPY --from=gst-wayland-cached-builder /artifacts/lib/ /usr/lib/

packages/scripts/entrypoint.sh

@@ -1,72 +1,176 @@
 #!/bin/bash
 set -euo pipefail
 
-# Wait for dbus socket to be ready
-echo "Waiting for DBus system bus socket..."
-DBUS_SOCKET="/run/dbus/system_bus_socket"
-for _ in {1..10}; do # Wait up to 10 seconds
-    if [ -e "$DBUS_SOCKET" ]; then
-        echo "DBus system bus socket is ready."
-        break
-    fi
-    sleep 1
-done
-if [ ! -e "$DBUS_SOCKET" ]; then
-    echo "Error: DBus system bus socket did not appear. Exiting."
-    exit 1
-fi
+# Configuration
+CACHE_DIR="/home/nestri/.cache/nvidia"
+NVIDIA_INSTALLER_DIR="/tmp"
+TIMEOUT_SECONDS=10
 
-# Wait for PipeWire to be ready
-echo "Waiting for PipeWire socket..."
-PIPEWIRE_SOCKET="/run/user/${UID}/pipewire-0"
-for _ in {1..10}; do # Wait up to 10 seconds
-    if [ -e "$PIPEWIRE_SOCKET" ]; then
-        echo "PipeWire socket is ready."
-        break
-    fi
-    sleep 1
-done
-if [ ! -e "$PIPEWIRE_SOCKET" ]; then
-    echo "Error: PipeWire socket did not appear. Exiting."
-    exit 1
-fi
+log() {
+    echo "[$(date +'%Y-%m-%d %H:%M:%S')] $1"
+}
 
-echo "Detecting GPU vendor..."
-source /etc/nestri/gpu_helpers.sh
+# Waits for a given socket to be ready
+wait_for_socket() {
+    local socket_path="$1"
+    local name="$2"
+    log "Waiting for $name socket at $socket_path..."
+    for ((i=1; i<=TIMEOUT_SECONDS; i++)); do
+        if [[ -e "$socket_path" ]]; then
+            log "$name socket is ready."
+            return 0
+        fi
+        sleep 1
+    done
+    log "Error: $name socket did not appear after ${TIMEOUT_SECONDS}s."
+    return 1
+}
 
-get_gpu_info
+# Ensures cache directory exists
+setup_cache() {
+    log "Setting up NVIDIA driver cache directory at $CACHE_DIR..."
+    mkdir -p "$CACHE_DIR" || {
+        log "Warning: Failed to create cache directory, continuing without cache."
+        return 1
+    }
+    chown nestri:nestri "$CACHE_DIR" 2>/dev/null || {
+        log "Warning: Failed to set cache directory ownership, continuing..."
+    }
+}
 
-# Check for NVIDIA so we can apply a workaround
-if [[ -n "${vendor_devices[nvidia]:-}" ]]; then
-    echo "NVIDIA GPU detected, applying driver fix..."
-    # Determine NVIDIA driver version from host
-    if [ -f "/proc/driver/nvidia/version" ]; then
-        NVIDIA_DRIVER_VERSION=$(head -n1 /proc/driver/nvidia/version | awk '{for(i=1;i<=NF;i++) if ($i ~ /^[0-9]+\.[0-9\.]+/) {print $i; exit}}')
-    elif command -v nvidia-smi &> /dev/null; then
-        NVIDIA_DRIVER_VERSION=$(nvidia-smi --version | grep -i 'driver version' | cut -d: -f2 | tr -d ' ')
-    else
-        echo "Failed to determine NVIDIA driver version. Exiting."
-        exit 1
+# Grabs NVIDIA driver installer
+get_nvidia_installer() {
+    local driver_version="$1"
+    local arch="$2"
+    local filename="NVIDIA-Linux-${arch}-${driver_version}.run"
+    local cached_file="${CACHE_DIR}/${filename}"
+    local tmp_file="${NVIDIA_INSTALLER_DIR}/${filename}"
+
+    # Check cache
+    if [[ -f "$cached_file" ]]; then
+        log "Found cached NVIDIA installer at $cached_file."
+        cp "$cached_file" "$tmp_file" || {
+            log "Warning: Failed to copy cached installer, proceeding with download."
+            rm -f "$cached_file" 2>/dev/null
+        }
     fi
 
-    NVIDIA_DRIVER_ARCH=$(uname -m)
-    filename="NVIDIA-Linux-${NVIDIA_DRIVER_ARCH}-${NVIDIA_DRIVER_VERSION}.run"
-
-    cd /tmp/
-    if [ ! -f "${filename}" ]; then
-        # Attempt multiple download sources
-        if ! wget "https://international.download.nvidia.com/XFree86/Linux-${NVIDIA_DRIVER_ARCH}/${NVIDIA_DRIVER_VERSION}/${filename}"; then
-            if ! wget "https://international.download.nvidia.com/tesla/${NVIDIA_DRIVER_VERSION}/${filename}"; then
-                echo "Failed to download NVIDIA driver from both XFree86 and Tesla repositories"
-                exit 1
+    # Download if not in tmp
+    if [[ ! -f "$tmp_file" ]]; then
+        log "Downloading NVIDIA driver installer ($filename)..."
+        local urls=(
+            "https://international.download.nvidia.com/XFree86/Linux-${arch}/${driver_version}/${filename}"
+            "https://international.download.nvidia.com/tesla/${driver_version}/${filename}"
+        )
+        local success=0
+        for url in "${urls[@]}"; do
+            if wget -q --show-progress "$url" -O "$tmp_file"; then
+                success=1
+                break
             fi
+            log "Failed to download from $url, trying next source..."
+        done
+
+        if [[ "$success" -eq 0 ]]; then
+            log "Error: Failed to download NVIDIA driver from all sources."
+            return 1
         fi
 
-        chmod +x "${filename}"
-        # Install driver components without kernel modules
-        sudo ./"${filename}" --silent --no-kernel-module --install-compat32-libs --no-nouveau-check --no-nvidia-modprobe --no-systemd --no-rpms --no-backup --no-check-for-alternate-installs
+        # Cache the downloaded file
+        cp "$tmp_file" "$cached_file" 2>/dev/null && \
+            chown nestri:nestri "$cached_file" 2>/dev/null || \
+            log "Warning: Failed to cache NVIDIA driver, continuing..."
     fi
-fi
 
-echo "Switching to nestri user for application startup..."
-exec sudo -E -u nestri /etc/nestri/entrypoint_nestri.sh
+    chmod +x "$tmp_file" || {
+        log "Error: Failed to make NVIDIA installer executable."
+        return 1
+    }
+    return 0
+}
+
+# Installs the NVIDIA driver
+install_nvidia_driver() {
+    local filename="$1"
+    log "Installing NVIDIA driver components from $filename..."
+    sudo ./"$filename" \
+        --silent \
+        --no-kernel-module \
+        --install-compat32-libs \
+        --no-nouveau-check \
+        --no-nvidia-modprobe \
+        --no-systemd \
+        --no-rpms \
+        --no-backup \
+        --no-check-for-alternate-installs || {
+        log "Error: NVIDIA driver installation failed."
+        return 1
+    }
+    log "NVIDIA driver installation completed."
+    return 0
+}
+
+main() {
+    # Wait for required sockets
+    wait_for_socket "/run/dbus/system_bus_socket" "DBus" || exit 1
+    wait_for_socket "/run/user/${UID}/pipewire-0" "PipeWire" || exit 1
+
+    # Load GPU helpers and detect GPU
+    log "Detecting GPU vendor..."
+    if [[ ! -f /etc/nestri/gpu_helpers.sh ]]; then
+        log "Error: GPU helpers script not found at /etc/nestri/gpu_helpers.sh."
+        exit 1
+    fi
+    source /etc/nestri/gpu_helpers.sh
+    get_gpu_info || {
+        log "Error: Failed to detect GPU information."
+        exit 1
+    }
+
+    # Handle NVIDIA GPU
+    if [[ -n "${vendor_devices[nvidia]:-}" ]]; then
+        log "NVIDIA GPU detected, applying driver fix..."
+
+        # Determine NVIDIA driver version
+        local nvidia_driver_version=""
+        if [[ -f "/proc/driver/nvidia/version" ]]; then
+            nvidia_driver_version=$(awk '/NVIDIA/ {for(i=1;i<=NF;i++) if ($i ~ /^[0-9]+\.[0-9\.]+/) {print $i; exit}}' /proc/driver/nvidia/version | head -n1)
+        elif command -v nvidia-smi >/dev/null 2>&1; then
+            nvidia_driver_version=$(nvidia-smi --version | grep -i 'driver version' | cut -d: -f2 | tr -d ' ')
+        fi
+
+        if [[ -z "$nvidia_driver_version" ]]; then
+            log "Error: Failed to determine NVIDIA driver version."
+            exit 1
+        fi
+        log "Detected NVIDIA driver version: $nvidia_driver_version"
+
+        # Set up cache and get installer
+        setup_cache
+        local arch=$(uname -m)
+        local filename="NVIDIA-Linux-${arch}-${nvidia_driver_version}.run"
+        cd "$NVIDIA_INSTALLER_DIR" || {
+            log "Error: Failed to change to $NVIDIA_INSTALLER_DIR."
+            exit 1
+        }
+        get_nvidia_installer "$nvidia_driver_version" "$arch" || exit 1
+
+        # Install driver
+        install_nvidia_driver "$filename" || exit 1
+    else
+        log "No NVIDIA GPU detected, skipping driver fix."
+    fi
+
+    # Switch to nestri user
+    log "Switching to nestri user for application startup..."
+    if [[ ! -x /etc/nestri/entrypoint_nestri.sh ]]; then
+        log "Error: Entry point script /etc/nestri/entrypoint_nestri.sh not found or not executable."
+        exit 1
+    fi
+    exec sudo -E -u nestri /etc/nestri/entrypoint_nestri.sh
+}
+
+# Trap signals for clean exit
+trap 'log "Received termination signal, exiting..."; exit 1' SIGINT SIGTERM
+
+main

packages/scripts/entrypoint_nestri.sh

@@ -1,222 +1,233 @@
 #!/bin/bash
 set -euo pipefail
 
-# Make user directory owned by the default user
-chown -f "$(id -nu):$(id -ng)" ~ || \
-      sudo chown -f "$(id -nu):$(id -ng)" ~ || \
-      chown -R -f -h --no-preserve-root "$(id -nu):$(id -ng)" ~ || \
-      sudo chown -R -f -h --no-preserve-root "$(id -nu):$(id -ng)" ~ || \
-      echo 'Failed to change user directory permissions, there may be permission issues'
+log() {
+    echo "[$(date +'%Y-%m-%d %H:%M:%S')] $1"
+}
 
-# Source environment variables from envs.sh
-if [ -f /etc/nestri/envs.sh ]; then
-    echo "Sourcing environment variables from envs.sh..."
-    source /etc/nestri/envs.sh
-else
-    echo "envs.sh not found! Ensure it exists at /etc/nestri/envs.sh."
-    exit 1
-fi
+# Ensures user directory ownership
+chown_user_directory() {
+    local user_group="$(id -nu):$(id -ng)"
+    chown -f "$user_group" ~ 2>/dev/null ||
+        sudo chown -f "$user_group" ~ 2>/dev/null ||
+        chown -R -f -h --no-preserve-root "$user_group" ~ 2>/dev/null ||
+        sudo chown -R -f -h --no-preserve-root "$user_group" ~ 2>/dev/null ||
+        log "Warning: Failed to change user directory permissions, there may be permission issues, continuing..."
+}
+
+# Parses resolution string
+parse_resolution() {
+    local resolution="$1"
+    if [[ -z "$resolution" ]]; then
+        log "Error: No resolution provided"
+        return 1
+    fi
+
+    IFS='x' read -r width height <<< "$resolution"
+    if ! [[ "$width" =~ ^[0-9]+$ ]] || ! [[ "$height" =~ ^[0-9]+$ ]]; then
+        log "Error: Invalid resolution format. Expected: WIDTHxHEIGHT (e.g., 1920x1080), got: $resolution"
+        return 1
+    fi
+
+    export WIDTH="$width"
+    export HEIGHT="$height"
+    return 0
+}
+
+# Loads environment variables
+load_envs() {
+    if [[ -f /etc/nestri/envs.sh ]]; then
+        log "Sourcing environment variables from envs.sh..."
+        source /etc/nestri/envs.sh
+    else
+        log "Error: envs.sh not found at /etc/nestri/envs.sh"
+        exit 1
+    fi
+}
 
 # Configuration
 MAX_RETRIES=3
+RETRY_COUNT=0
 
-# Helper function to restart the chain
-restart_chain() {
-    echo "Restarting nestri-server, compositor..."
-
-    # Start nestri-server
-    start_nestri_server
-    RETRY_COUNT=0
+# Kills process if running
+kill_if_running() {
+    local pid="$1"
+    local name="$2"
+    if [[ -n "$pid" ]] && kill -0 "$pid" 2>/dev/null; then
+        log "Killing existing $name process (PID: $pid)..."
+        kill "$pid"
+        wait "$pid" 2>/dev/null || true
+    fi
 }
 
-# Function to start nestri-server
+# Starts nestri-server
 start_nestri_server() {
-    if [[ -n "${NESTRI_PID:-}" ]] && kill -0 "${NESTRI_PID}" 2 >/dev/null; then
-        echo "Killing existing nestri-server process..."
-        kill "${NESTRI_PID}"
-    fi
+    kill_if_running "${NESTRI_PID:-}" "nestri-server"
 
-    echo "Starting nestri-server..."
-    nestri-server $(echo $NESTRI_PARAMS) &
+    log "Starting nestri-server..."
+    nestri-server $NESTRI_PARAMS &
     NESTRI_PID=$!
 
-    # Wait for Wayland display (wayland-1) to be ready
-    echo "Waiting for Wayland display 'wayland-1' to be ready..."
+    log "Waiting for Wayland display 'wayland-1'..."
     WAYLAND_SOCKET="${XDG_RUNTIME_DIR}/wayland-1"
-    for _ in {1..15}; do # Wait up to 15 seconds
-        if [ -e "$WAYLAND_SOCKET" ]; then
-            echo "Wayland display 'wayland-1' is ready."
-            sleep 5  # necessary sleep - reduces chance that non-ready socket is used
+    for ((i=1; i<=15; i++)); do
+        if [[ -e "$WAYLAND_SOCKET" ]]; then
+            log "Wayland display 'wayland-1' ready."
+            sleep 3
             start_compositor
             return
         fi
         sleep 1
     done
 
-    echo "Error: Wayland display 'wayland-1' did not appear. Incrementing retry count..."
-    ((RETRY_COUNT++))
-    if [ "$RETRY_COUNT" -ge "$MAX_RETRIES" ]; then
-        echo "Max retries reached for nestri-server. Exiting."
-        exit 1
-    fi
+    log "Error: Wayland display 'wayland-1' not available."
+    increment_retry "nestri-server"
     restart_chain
 }
 
-# Function to start compositor (labwc)
+# Starts compositor (labwc)
 start_compositor() {
-    if [[ -n "${COMPOSITOR_PID:-}" ]] && kill -0 "${COMPOSITOR_PID}" 2 >/dev/null; then
-        echo "Killing existing compositor process..."
-        kill "${COMPOSITOR_PID}"
-    fi
+    kill_if_running "${COMPOSITOR_PID:-}" "compositor"
 
-    echo "Pre-configuring compositor..."
+    log "Pre-configuring compositor..."
     mkdir -p "${HOME}/.config/labwc/"
-    echo '<?xml version="1.0" encoding="UTF-8"?><labwc_config><keyboard><default/></keyboard><mouse><default/><context name="Root"><mousebind button="Left" action="Press"/><mousebind button="Right" action="Press"/><mousebind button="Middle" action="Press"/></context></mouse></labwc_config>' > ~/.config/labwc/rc.xml
+    cat > ~/.config/labwc/rc.xml << 'EOF'
+<?xml version="1.0" encoding="UTF-8"?>
+<labwc_config>
+    <keyboard><default/></keyboard>
+    <mouse><default/>
+        <context name="Root">
+            <mousebind button="Left" action="Press"/>
+            <mousebind button="Right" action="Press"/>
+            <mousebind button="Middle" action="Press"/>
+        </context>
+    </mouse>
+</labwc_config>
+EOF
     echo '<?xml version="1.0" encoding="UTF-8"?><openbox_menu></openbox_menu>' > ~/.config/labwc/menu.xml
 
-    echo "Starting compositor..."
+    log "Starting compositor..."
     rm -rf /tmp/.X11-unix && mkdir -p /tmp/.X11-unix && chown nestri:nestri /tmp/.X11-unix
     WAYLAND_DISPLAY=wayland-1 WLR_BACKENDS=wayland labwc &
     COMPOSITOR_PID=$!
 
-    # Wait for compositor to initialize
-    echo "Waiting for compositor to initialize..."
+    log "Waiting for compositor to initialize..."
     COMPOSITOR_SOCKET="${XDG_RUNTIME_DIR}/wayland-0"
-    for _ in {1..15}; do
-        if [ -e "$COMPOSITOR_SOCKET" ]; then
-            echo "compositor is initialized, wayland-0 output ready."
-            sleep 3  # necessary sleep - reduces chance that non-ready socket is used
+    for ((i=1; i<=15; i++)); do
+        if [[ -e "$COMPOSITOR_SOCKET" ]]; then
+            log "Compositor initialized, wayland-0 ready."
+            sleep 2
             start_wlr_randr
             return
         fi
         sleep 1
     done
 
-    echo "Error: compositor did not initialize correctly. Incrementing retry count..."
-    ((RETRY_COUNT++))
-    if [ "$RETRY_COUNT" -ge "$MAX_RETRIES" ]; then
-        echo "Max retries reached for compositor. Exiting."
-        exit 1
-    fi
+    log "Error: Compositor did not initialize."
+    increment_retry "compositor"
     start_compositor
 }
 
-# Function to run wlr-randr
+# Configures resolution with wlr-randr
 start_wlr_randr() {
-    echo "Configuring resolution with wlr-randr..."
+    log "Configuring resolution with wlr-randr..."
     OUTPUT_NAME=$(WAYLAND_DISPLAY=wayland-0 wlr-randr --json | jq -r '.[] | select(.enabled == true) | .name' | head -n 1)
-    if [ -z "$OUTPUT_NAME" ]; then
-        echo "Error: No enabled outputs detected, exiting."
+    if [[ -z "$OUTPUT_NAME" ]]; then
+        log "Error: No enabled outputs detected."
         exit 1
     fi
 
-    # Retry logic for wlr-randr
     local WLR_RETRIES=0
     while ! WAYLAND_DISPLAY=wayland-0 wlr-randr --output "$OUTPUT_NAME" --custom-mode "$RESOLUTION"; do
-        echo "Error: Failed to configure wlr-randr. Retrying..."
+        log "Error: Failed to configure wlr-randr. Retrying..."
         ((WLR_RETRIES++))
-        if [ "$WLR_RETRIES" -ge "$MAX_RETRIES" ]; then
-            echo "Max retries reached for wlr-randr, exiting."
+        if [[ "$WLR_RETRIES" -ge "$MAX_RETRIES" ]]; then
+            log "Error: Max retries reached for wlr-randr."
             exit 1
         fi
         sleep 2
     done
-    echo "wlr-randr configuration successful."
-    sleep 2  # necessary sleep - makes sure resolution is changed before next step(s)
+    log "wlr-randr configuration successful."
+    sleep 2
 }
 
-# Function to start Steam
+# Starts Steam
 start_steam() {
-    if [[ -n "${STEAM_PID:-}" ]] && kill -0 "${STEAM_PID}" 2 >/dev/null; then
-        echo "Killing existing Steam process..."
-        kill "${STEAM_PID}"
-    fi
+    kill_if_running "${STEAM_PID:-}" "Steam"
 
-    echo "Starting Steam with -tenfoot..."
-    WAYLAND_DISPLAY=wayland-0 steam-native -tenfoot &
+    log "Starting Steam with -tenfoot..."
+    steam-native -tenfoot &
     STEAM_PID=$!
 
-    # Verify Steam started successfully
     sleep 2
     if ! kill -0 "$STEAM_PID" 2>/dev/null; then
-        echo "Error: Steam failed to start."
+        log "Error: Steam failed to start."
         return 1
     fi
-    echo "Steam started successfully."
+    log "Steam started successfully."
     return 0
 }
 
-# Main loop to monitor processes
+# Increments retry counter
+increment_retry() {
+    local component="$1"
+    ((RETRY_COUNT++))
+    if [[ "$RETRY_COUNT" -ge "$MAX_RETRIES" ]]; then
+        log "Error: Max retries reached for $component."
+        exit 1
+    fi
+}
+
+# Restarts the chain
+restart_chain() {
+    log "Restarting nestri-server and compositor..."
+    RETRY_COUNT=0
+    start_nestri_server
+}
+
+# Cleans up processes
+cleanup() {
+    log "Terminating processes..."
+    kill_if_running "${NESTRI_PID:-}" "nestri-server"
+    kill_if_running "${COMPOSITOR_PID:-}" "compositor"
+    kill_if_running "${STEAM_PID:-}" "Steam"
+    exit 0
+}
+
+# Monitor processes for unexpected exits
 main_loop() {
-    trap 'echo "Terminating...";
-if [[ -n "${NESTRI_PID:-}" ]] && kill -0 "${NESTRI_PID}" 2>/dev/null; then
-kill "${NESTRI_PID}"
-fi
-if [[ -n "${COMPOSITOR_PID:-}" ]] && kill -0 "${COMPOSITOR_PID}" 2>/dev/null; then
-kill "${COMPOSITOR_PID}"
-fi
-if [[ -n "${STEAM_PID:-}" ]] && kill -0 "${STEAM_PID}" 2>/dev/null; then
-kill "${STEAM_PID}"
-fi
-    exit 0' SIGINT SIGTERM
+    trap cleanup SIGINT SIGTERM
 
     while true; do
-        # Wait for any child process to exit
-        wait -n
-
-        # Check which process exited
-        if ! kill -0 ${NESTRI_PID:-} 2 >/dev/null; then
-            echo "nestri-server crashed. Restarting chain..."
-            ((RETRY_COUNT++))
-            if [ "$RETRY_COUNT" -ge "$MAX_RETRIES" ]; then
-                echo "Max retries reached for nestri-server. Exiting."
-                exit 1
-            fi
+        sleep 1
+        # Check nestri-server
+        if [[ -n "${NESTRI_PID:-}" ]] && ! kill -0 "${NESTRI_PID}" 2>/dev/null; then
+            log "nestri-server died."
+            increment_retry "nestri-server"
             restart_chain
-            start_steam || {
-                echo "Failed to restart Steam after chain restart."
-                ((RETRY_COUNT++))
-                if [ "$RETRY_COUNT" -ge "$MAX_RETRIES" ]; then
-                    echo "Max retries reached. Exiting."
-                    exit 1
-                fi
-            }
-        elif ! kill -0 ${COMPOSITOR_PID:-} 2 >/dev/null; then
-            echo "compositor crashed. Restarting compositor..."
-            ((RETRY_COUNT++))
-            if [ "$RETRY_COUNT" -ge "$MAX_RETRIES" ]; then
-                echo "Max retries reached for compositor. Exiting."
-                exit 1
-            fi
+            start_steam || increment_retry "Steam"
+        # Check compositor
+        elif [[ -n "${COMPOSITOR_PID:-}" ]] && ! kill -0 "${COMPOSITOR_PID}" 2>/dev/null; then
+            log "compositor died."
+            increment_retry "compositor"
             start_compositor
-            start_steam || {
-                echo "Failed to restart Steam after compositor restart."
-                ((RETRY_COUNT++))
-                if [ "$RETRY_COUNT" -ge "$MAX_RETRIES" ]; then
-                    echo "Max retries reached. Exiting."
-                    exit 1
-                fi
-            }
-        elif ! kill -0 ${STEAM_PID:-} 2 >/dev/null; then
-            echo "Steam crashed. Restarting Steam..."
-            ((RETRY_COUNT++))
-            if [ "$RETRY_COUNT" -ge "$MAX_RETRIES" ]; then
-                echo "Max retries reached for Steam. Exiting."
-                exit 1
-            fi
-            start_steam
+            start_steam || increment_retry "Steam"
+        # Check Steam
+        elif [[ -n "${STEAM_PID:-}" ]] && ! kill -0 "${STEAM_PID}" 2>/dev/null; then
+            log "Steam died."
+            increment_retry "Steam"
+            start_steam || increment_retry "Steam"
         fi
     done
 }
 
-# Initialize retry counter
-RETRY_COUNT=0
+main() {
+    chown_user_directory
+    load_envs
+    #parse_resolution "${RESOLUTION:-1920x1080}" || exit 1 # Not used currently
+    restart_chain
+    start_steam || increment_retry "Steam"
+    main_loop
+}
 
-# Start the initial chain
-restart_chain
-
-# Start Steam after initial setup
-start_steam
-
-# Enter monitoring loop
-main_loop
+main

packages/scripts/envs.sh

@@ -7,5 +7,8 @@ export XDG_SESSION_TYPE=wayland
 export DISPLAY=:0
 export $(dbus-launch)
 
+# Causes some setups to break
+export PROTON_NO_FSYNC=1
+
 # Our preferred prefix
 export WINEPREFIX=/home/${USER}/.nestripfx/

packages/scripts/supervisord.conf

@@ -27,6 +27,7 @@ autorestart=true
 autostart=true
 startretries=3
 priority=3
+nice=-10
 
 [program:pipewire-pulse]
 user=nestri
@@ -35,6 +36,7 @@ autorestart=true
 autostart=true
 startretries=3
 priority=4
+nice=-10
 
 [program:wireplumber]
 user=nestri
@@ -43,6 +45,7 @@ autorestart=true
 autostart=true
 startretries=3
 priority=5
+nice=-10
 
 [program:entrypoint]
 user=root

packages/server/Cargo.toml

@@ -8,22 +8,24 @@ name = "nestri-server"
 path = "src/main.rs"
 
 [dependencies]
-gst = { package = "gstreamer", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", branch = "main", features = ["v1_24"] }
-gst-webrtc = { package = "gstreamer-webrtc", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", branch = "main", features = ["v1_24"] }
+gst = { package = "gstreamer", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", branch = "main", features = ["v1_26"] }
+gst-webrtc = { package = "gstreamer-webrtc", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", branch = "main", features = ["v1_26"] }
 gstrswebrtc = { package = "gst-plugin-webrtc", git = "https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs", branch = "main", features = ["v1_22"] }
-serde = {version = "1.0.214", features = ["derive"] }
-tokio = { version = "1.41.0", features = ["full"] }
-clap = { version = "4.5.20", features = ["env"] }
-serde_json = "1.0.132"
-webrtc = "0.12.0"
-regex = "1.11.1"
-rand = "0.9.0"
-rustls = { version = "0.23.17", features = ["ring"] }
-tokio-tungstenite = { version = "0.26.1", features = ["native-tls"] }
-log = { version = "0.4.22", features = ["std"] }
-chrono = "0.4.38"
-futures-util = "0.3.31"
-num-derive = "0.4.2"
-num-traits = "0.2.19"
-prost = "0.13.4"
-prost-types = "0.13.4"
+serde = {version = "1.0", features = ["derive"] }
+tokio = { version = "1.44", features = ["full"] }
+clap = { version = "4.5", features = ["env"] }
+serde_json = "1.0"
+webrtc = "0.12"
+regex = "1.11"
+rand = "0.9"
+rustls = { version = "0.23", features = ["ring"] }
+tokio-tungstenite = { version = "0.26", features = ["native-tls"] }
+log = { version = "0.4", features = ["std"] }
+chrono = "0.4"
+futures-util = "0.3"
+num-derive = "0.4"
+num-traits = "0.2"
+prost = "0.13"
+prost-types = "0.13"
+parking_lot = "0.12"
+atomic_refcell = "0.1"

packages/server/src/args.rs

@@ -22,19 +22,11 @@ impl Args {
                     .default_value("false"),
             )
             .arg(
-                Arg::new("debug-feed")
+                Arg::new("debug")
                     .short('d')
-                    .long("debug-feed")
-                    .env("DEBUG_FEED")
-                    .help("Debug by showing a window on host")
-                    .default_value("false"),
-            )
-            .arg(
-                Arg::new("debug-latency")
-                    .short('l')
-                    .long("debug-latency")
-                    .env("DEBUG_LATENCY")
-                    .help("Debug latency by showing time on feed")
+                    .long("debug")
+                    .env("DEBUG")
+                    .help("Enable additional debugging information and features")
                     .default_value("false"),
             )
             .arg(

packages/server/src/args/app_args.rs

@@ -1,10 +1,8 @@
 pub struct AppArgs {
     /// Verbose output mode
     pub verbose: bool,
-    /// Debug the pipeline by showing a window on host
-    pub debug_feed: bool,
-    /// Debug the latency by showing time in stream
-    pub debug_latency: bool,
+    /// Enable additional debug information and features, may affect performance
+    pub debug: bool,
 
     /// Virtual display resolution
     pub resolution: (u32, u32),
@@ -24,10 +22,8 @@ impl AppArgs {
         Self {
             verbose: matches.get_one::<String>("verbose").unwrap() == "true"
                 || matches.get_one::<String>("verbose").unwrap() == "1",
-            debug_feed: matches.get_one::<String>("debug-feed").unwrap() == "true"
-                || matches.get_one::<String>("debug-feed").unwrap() == "1",
-            debug_latency: matches.get_one::<String>("debug-latency").unwrap() == "true"
-                || matches.get_one::<String>("debug-latency").unwrap() == "1",
+            debug: matches.get_one::<String>("debug").unwrap() == "true"
+                || matches.get_one::<String>("debug").unwrap() == "1",
             resolution: {
                 let res = matches
                     .get_one::<String>("resolution")
@@ -65,8 +61,7 @@ impl AppArgs {
     pub fn debug_print(&self) {
         println!("AppArgs:");
         println!("> verbose: {}", self.verbose);
-        println!("> debug_feed: {}", self.debug_feed);
-        println!("> debug_latency: {}", self.debug_latency);
+        println!("> debug: {}", self.debug);
         println!("> resolution: {}x{}", self.resolution.0, self.resolution.1);
         println!("> framerate: {}", self.framerate);
         println!("> relay_url: {}", self.relay_url);

packages/server/src/enc_helper.rs

@@ -249,8 +249,10 @@ pub fn encoder_gop_params(encoder: &VideoEncoderInfo, gop_size: u32) -> VideoEnc
 pub fn encoder_low_latency_params(
     encoder: &VideoEncoderInfo,
     rate_control: &RateControl,
+    framerate: u32,
 ) -> VideoEncoderInfo {
-    let mut encoder_optz = encoder_gop_params(encoder, 30);
+    // 2 second GOP size, maybe lower to 1 second for fast recovery, if needed?
+    let mut encoder_optz = encoder_gop_params(encoder, framerate * 2);
 
     match encoder_optz.encoder_api {
         EncoderAPI::QSV => {

packages/server/src/main.rs

@@ -128,8 +128,11 @@ fn handle_encoder_video_settings(
     args: &args::Args,
     video_encoder: &enc_helper::VideoEncoderInfo,
 ) -> enc_helper::VideoEncoderInfo {
-    let mut optimized_encoder =
-        enc_helper::encoder_low_latency_params(&video_encoder, &args.encoding.video.rate_control);
+    let mut optimized_encoder = enc_helper::encoder_low_latency_params(
+        &video_encoder,
+        &args.encoding.video.rate_control,
+        args.app.framerate,
+    );
     // Handle rate-control method
     match &args.encoding.video.rate_control {
         encoding_args::RateControl::CQP(cqp) => {
@@ -235,7 +238,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
         encoding_args::AudioCaptureMethod::PipeWire => {
             gst::ElementFactory::make("pipewiresrc").build()?
         }
-        _ => gst::ElementFactory::make("alsasrc").build()?,
+        encoding_args::AudioCaptureMethod::ALSA => gst::ElementFactory::make("alsasrc").build()?,
     };
 
     // Audio Converter Element
@@ -259,6 +262,10 @@ async fn main() -> Result<(), Box<dyn Error>> {
             _ => 128000i32,
         },
     );
+    // If has "frame-size" (opus), set to 10 for lower latency (below 10 seems to be too low?)
+    if audio_encoder.has_property("frame-size") {
+        audio_encoder.set_property_from_str("frame-size", "10");
+    }
 
     /* Video */
     // Video Source Element
@@ -299,6 +306,18 @@ async fn main() -> Result<(), Box<dyn Error>> {
     let video_encoder = gst::ElementFactory::make(video_encoder_info.name.as_str()).build()?;
     video_encoder_info.apply_parameters(&video_encoder, args.app.verbose);
 
+    // Video parser Element, required for GStreamer 1.26 as it broke some things..
+    let video_parser;
+    if video_encoder_info.codec == enc_helper::VideoCodec::H264 {
+        video_parser = Some(
+            gst::ElementFactory::make("h264parse")
+                .property("config-interval", -1i32)
+                .build()?,
+        );
+    } else {
+        video_parser = None;
+    }
+
     /* Output */
     // WebRTC sink Element
     let signaller = NestriSignaller::new(nestri_ws.clone(), pipeline.clone());
@@ -307,20 +326,50 @@ async fn main() -> Result<(), Box<dyn Error>> {
     webrtcsink.set_property_from_str("congestion-control", "disabled");
     webrtcsink.set_property("do-retransmission", false);
 
+    /* Queues */
+    let video_queue = gst::ElementFactory::make("queue2")
+        .property("max-size-buffers", 3u32)
+        .property("max-size-time", 0u64)
+        .property("max-size-bytes", 0u32)
+        .build()?;
+
+    let audio_queue = gst::ElementFactory::make("queue2")
+        .property("max-size-buffers", 3u32)
+        .property("max-size-time", 0u64)
+        .property("max-size-bytes", 0u32)
+        .build()?;
+
+    /* Clock Sync */
+    let video_clocksync = gst::ElementFactory::make("clocksync")
+        .property("sync-to-first", true)
+        .build()?;
+
+    let audio_clocksync = gst::ElementFactory::make("clocksync")
+        .property("sync-to-first", true)
+        .build()?;
+
     // Add elements to the pipeline
     pipeline.add_many(&[
         webrtcsink.upcast_ref(),
         &video_encoder,
         &video_converter,
         &caps_filter,
+        &video_queue,
+        &video_clocksync,
         &video_source,
         &audio_encoder,
         &audio_capsfilter,
+        &audio_queue,
+        &audio_clocksync,
         &audio_rate,
         &audio_converter,
         &audio_source,
     ])?;
 
+    if let Some(parser) = &video_parser {
+        pipeline.add(parser)?;
+    }
+
     // If DMA-BUF is enabled, add glupload, color conversion and caps filter
     if args.app.dma_buf {
         pipeline.add_many(&[&glupload, &glcolorconvert, &gl_caps_filter])?;
@@ -332,6 +381,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
         &audio_converter,
         &audio_rate,
         &audio_capsfilter,
+        &audio_queue,
+        &audio_clocksync,
         &audio_encoder,
         webrtcsink.upcast_ref(),
     ])?;
@@ -342,30 +393,45 @@ async fn main() -> Result<(), Box<dyn Error>> {
         gst::Element::link_many(&[
             &video_source,
             &caps_filter,
+            &video_queue,
+            &video_clocksync,
             &glupload,
             &glcolorconvert,
             &gl_caps_filter,
             &video_encoder,
-            webrtcsink.upcast_ref(),
         ])?;
     } else {
         // Link video source to caps_filter, video_converter, video_encoder, webrtcsink
         gst::Element::link_many(&[
             &video_source,
             &caps_filter,
+            &video_queue,
+            &video_clocksync,
             &video_converter,
             &video_encoder,
-            webrtcsink.upcast_ref(),
         ])?;
     }
 
+    // Link video parser if present with webrtcsink, otherwise just link webrtc sink
+    if let Some(parser) = &video_parser {
+        gst::Element::link_many(&[&video_encoder, parser, webrtcsink.upcast_ref()])?;
+    } else {
+        gst::Element::link_many(&[&video_encoder, webrtcsink.upcast_ref()])?;
+    }
+
+    // Set QOS
+    video_encoder.set_property("qos", true);
+
     // Optimize latency of pipeline
     video_source
         .sync_state_with_parent()
         .expect("failed to sync with parent");
     video_source.set_property("do-timestamp", &true);
     audio_source.set_property("do-timestamp", &true);
+
     pipeline.set_property("latency", &0u64);
+    pipeline.set_property("async-handling", true);
+    pipeline.set_property("message-forward", true);
 
     // Run both pipeline and websocket tasks concurrently
     let result = run_pipeline(pipeline.clone()).await;

packages/server/src/nestrisink/imp.rs

@@ -7,47 +7,53 @@ use crate::proto::proto::proto_input::InputType::{
 };
 use crate::proto::proto::{ProtoInput, ProtoMessageInput};
 use crate::websocket::NestriWebSocket;
+use atomic_refcell::AtomicRefCell;
 use glib::subclass::prelude::*;
 use gst::glib;
 use gst::prelude::*;
 use gst_webrtc::{WebRTCSDPType, WebRTCSessionDescription, gst_sdp};
 use gstrswebrtc::signaller::{Signallable, SignallableImpl};
+use parking_lot::RwLock as PLRwLock;
 use prost::Message;
-use std::collections::HashSet;
 use std::sync::{Arc, LazyLock};
-use std::sync::{Mutex, RwLock};
 use webrtc::ice_transport::ice_candidate::RTCIceCandidateInit;
 use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
 
 pub struct Signaller {
-    nestri_ws: RwLock<Option<Arc<NestriWebSocket>>>,
-    pipeline: RwLock<Option<Arc<gst::Pipeline>>>,
-    data_channel: RwLock<Option<gst_webrtc::WebRTCDataChannel>>,
+    nestri_ws: PLRwLock<Option<Arc<NestriWebSocket>>>,
+    pipeline: PLRwLock<Option<Arc<gst::Pipeline>>>,
+    data_channel: AtomicRefCell<Option<gst_webrtc::WebRTCDataChannel>>,
 }
 impl Default for Signaller {
     fn default() -> Self {
         Self {
-            nestri_ws: RwLock::new(None),
-            pipeline: RwLock::new(None),
-            data_channel: RwLock::new(None),
+            nestri_ws: PLRwLock::new(None),
+            pipeline: PLRwLock::new(None),
+            data_channel: AtomicRefCell::new(None),
         }
     }
 }
 impl Signaller {
     pub fn set_nestri_ws(&self, nestri_ws: Arc<NestriWebSocket>) {
-        *self.nestri_ws.write().unwrap() = Some(nestri_ws);
+        *self.nestri_ws.write() = Some(nestri_ws);
     }
 
     pub fn set_pipeline(&self, pipeline: Arc<gst::Pipeline>) {
-        *self.pipeline.write().unwrap() = Some(pipeline);
+        *self.pipeline.write() = Some(pipeline);
     }
 
     pub fn get_pipeline(&self) -> Option<Arc<gst::Pipeline>> {
-        self.pipeline.read().unwrap().clone()
+        self.pipeline.read().clone()
     }
 
     pub fn set_data_channel(&self, data_channel: gst_webrtc::WebRTCDataChannel) {
-        *self.data_channel.write().unwrap() = Some(data_channel);
+        match self.data_channel.try_borrow_mut() {
+            Ok(mut dc) => *dc = Some(data_channel),
+            Err(_) => gst::warning!(
+                gst::CAT_DEFAULT,
+                "Failed to set data channel - already borrowed"
+            ),
+        }
     }
 
     /// Helper method to clean things up
@@ -55,7 +61,6 @@ impl Signaller {
         let nestri_ws = {
             self.nestri_ws
                 .read()
-                .unwrap()
                 .clone()
                 .expect("NestriWebSocket not set")
         };
@@ -145,8 +150,9 @@ impl Signaller {
                                 &"nestri-data-channel",
                                 &gst::Structure::builder("config")
                                     .field("ordered", &true)
-                                    .field("max-retransmits", &0u32)
+                                    .field("max-retransmits", &2u32)
                                     .field("priority", "high")
+                                    .field("protocol", "raw")
                                     .build(),
                             ],
                         ),
@@ -175,7 +181,6 @@ impl SignallableImpl for Signaller {
         let nestri_ws = {
             self.nestri_ws
                 .read()
-                .unwrap()
                 .clone()
                 .expect("NestriWebSocket not set")
         };
@@ -266,7 +271,6 @@ impl SignallableImpl for Signaller {
         let nestri_ws = {
             self.nestri_ws
                 .read()
-                .unwrap()
                 .clone()
                 .expect("NestriWebSocket not set")
         };
@@ -297,7 +301,6 @@ impl SignallableImpl for Signaller {
         let nestri_ws = {
             self.nestri_ws
                 .read()
-                .unwrap()
                 .clone()
                 .expect("NestriWebSocket not set")
         };
@@ -360,9 +363,6 @@ impl ObjectImpl for Signaller {
 
 fn setup_data_channel(data_channel: &gst_webrtc::WebRTCDataChannel, pipeline: &gst::Pipeline) {
     let pipeline = pipeline.clone();
-    // A shared state to track currently pressed keys
-    let pressed_keys = Arc::new(Mutex::new(HashSet::new()));
-    let pressed_buttons = Arc::new(Mutex::new(HashSet::new()));
 
     data_channel.connect_on_message_data(move |_data_channel, data| {
         if let Some(data) = data {
@@ -370,9 +370,7 @@ fn setup_data_channel(data_channel: &gst_webrtc::WebRTCDataChannel, pipeline: &g
                 Ok(message_input) => {
                     if let Some(input_msg) = message_input.data {
                         // Process the input message and create an event
-                        if let Some(event) =
-                            handle_input_message(input_msg, &pressed_keys, &pressed_buttons)
-                        {
+                        if let Some(event) = handle_input_message(input_msg) {
                             // Send the event to pipeline, result bool is ignored
                             let _ = pipeline.send_event(event);
                         }
@@ -388,11 +386,7 @@ fn setup_data_channel(data_channel: &gst_webrtc::WebRTCDataChannel, pipeline: &g
     });
 }
 
-fn handle_input_message(
-    input_msg: ProtoInput,
-    pressed_keys: &Arc<Mutex<HashSet<i32>>>,
-    pressed_buttons: &Arc<Mutex<HashSet<i32>>>,
-) -> Option<gst::Event> {
+fn handle_input_message(input_msg: ProtoInput) -> Option<gst::Event> {
     if let Some(input_type) = input_msg.input_type {
         match input_type {
             MouseMove(data) => {
@@ -412,13 +406,6 @@ fn handle_input_message(
                 Some(gst::event::CustomUpstream::new(structure))
             }
             KeyDown(data) => {
-                let mut keys = pressed_keys.lock().unwrap();
-                // If the key is already pressed, return to prevent key lockup
-                if keys.contains(&data.key) {
-                    return None;
-                }
-                keys.insert(data.key);
-
                 let structure = gst::Structure::builder("KeyboardKey")
                     .field("key", data.key as u32)
                     .field("pressed", true)
@@ -427,10 +414,6 @@ fn handle_input_message(
                 Some(gst::event::CustomUpstream::new(structure))
             }
             KeyUp(data) => {
-                let mut keys = pressed_keys.lock().unwrap();
-                // Remove the key from the pressed state when released
-                keys.remove(&data.key);
-
                 let structure = gst::Structure::builder("KeyboardKey")
                     .field("key", data.key as u32)
                     .field("pressed", false)
@@ -447,13 +430,6 @@ fn handle_input_message(
                 Some(gst::event::CustomUpstream::new(structure))
             }
             MouseKeyDown(data) => {
-                let mut buttons = pressed_buttons.lock().unwrap();
-                // If the button is already pressed, return to prevent button lockup
-                if buttons.contains(&data.key) {
-                    return None;
-                }
-                buttons.insert(data.key);
-
                 let structure = gst::Structure::builder("MouseButton")
                     .field("button", data.key as u32)
                     .field("pressed", true)
@@ -462,10 +438,6 @@ fn handle_input_message(
                 Some(gst::event::CustomUpstream::new(structure))
             }
             MouseKeyUp(data) => {
-                let mut buttons = pressed_buttons.lock().unwrap();
-                // Remove the button from the pressed state when released
-                buttons.remove(&data.key);
-
                 let structure = gst::Structure::builder("MouseButton")
                     .field("button", data.key as u32)
                     .field("pressed", false)

packages/server/src/proto/gen.rs

@@ -0,0 +1,202 @@
+// @generated
+// This file is @generated by prost-build.
+/// EntityState represents the state of an entity in the mesh (e.g., a room).
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct EntityState {
+    /// Type of entity (e.g., "room")
+    #[prost(string, tag="1")]
+    pub entity_type: ::prost::alloc::string::String,
+    /// Unique identifier (e.g., room name)
+    #[prost(string, tag="2")]
+    pub entity_id: ::prost::alloc::string::String,
+    /// Whether the entity is active
+    #[prost(bool, tag="3")]
+    pub active: bool,
+    /// Relay ID that owns this entity
+    #[prost(string, tag="4")]
+    pub owner_relay_id: ::prost::alloc::string::String,
+}
+/// MeshMessage is the top-level message for all relay-to-relay communication.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct MeshMessage {
+    #[prost(oneof="mesh_message::Type", tags="1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13")]
+    pub r#type: ::core::option::Option<mesh_message::Type>,
+}
+/// Nested message and enum types in `MeshMessage`.
+pub mod mesh_message {
+    #[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Oneof)]
+    pub enum Type {
+        /// Level 0
+        #[prost(message, tag="1")]
+        StateUpdate(super::StateUpdate),
+        #[prost(message, tag="2")]
+        Ack(super::Ack),
+        #[prost(message, tag="3")]
+        RetransmissionRequest(super::RetransmissionRequest),
+        #[prost(message, tag="4")]
+        Retransmission(super::Retransmission),
+        #[prost(message, tag="5")]
+        Heartbeat(super::Heartbeat),
+        #[prost(message, tag="6")]
+        SuspectRelay(super::SuspectRelay),
+        #[prost(message, tag="7")]
+        Disconnect(super::Disconnect),
+        /// Level 1
+        #[prost(message, tag="8")]
+        ForwardSdp(super::ForwardSdp),
+        #[prost(message, tag="9")]
+        ForwardIce(super::ForwardIce),
+        #[prost(message, tag="10")]
+        ForwardIngest(super::ForwardIngest),
+        #[prost(message, tag="11")]
+        StreamRequest(super::StreamRequest),
+        /// Level 2
+        #[prost(message, tag="12")]
+        Handshake(super::Handshake),
+        #[prost(message, tag="13")]
+        HandshakeResponse(super::HandshakeResponse),
+    }
+}
+/// Handshake to inititiate new connection to mesh.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Handshake {
+    /// UUID of the relay
+    #[prost(string, tag="1")]
+    pub relay_id: ::prost::alloc::string::String,
+    /// base64 encoded Diffie-Hellman public key
+    #[prost(string, tag="2")]
+    pub dh_public_key: ::prost::alloc::string::String,
+}
+/// HandshakeResponse to respond to a mesh joiner.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct HandshakeResponse {
+    #[prost(string, tag="1")]
+    pub relay_id: ::prost::alloc::string::String,
+    #[prost(string, tag="2")]
+    pub dh_public_key: ::prost::alloc::string::String,
+    /// relay id to signature
+    #[prost(map="string, string", tag="3")]
+    pub approvals: ::std::collections::HashMap<::prost::alloc::string::String, ::prost::alloc::string::String>,
+}
+/// Forwarded SDP from another relay.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ForwardSdp {
+    #[prost(string, tag="1")]
+    pub room_name: ::prost::alloc::string::String,
+    #[prost(string, tag="2")]
+    pub participant_id: ::prost::alloc::string::String,
+    #[prost(string, tag="3")]
+    pub sdp: ::prost::alloc::string::String,
+    /// "offer" or "answer"
+    #[prost(string, tag="4")]
+    pub r#type: ::prost::alloc::string::String,
+}
+/// Forwarded ICE candidate from another relay.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ForwardIce {
+    #[prost(string, tag="1")]
+    pub room_name: ::prost::alloc::string::String,
+    #[prost(string, tag="2")]
+    pub participant_id: ::prost::alloc::string::String,
+    #[prost(string, tag="3")]
+    pub candidate: ::prost::alloc::string::String,
+}
+/// Forwarded ingest room from another relay.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ForwardIngest {
+    #[prost(string, tag="1")]
+    pub room_name: ::prost::alloc::string::String,
+}
+/// Stream request from mesh.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct StreamRequest {
+    #[prost(string, tag="1")]
+    pub room_name: ::prost::alloc::string::String,
+}
+/// StateUpdate propagates entity state changes across the mesh.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct StateUpdate {
+    /// Unique sequence number for this update
+    #[prost(uint64, tag="1")]
+    pub sequence_number: u64,
+    /// Key: entity_id (e.g., room name), Value: EntityState
+    #[prost(map="string, message", tag="2")]
+    pub entities: ::std::collections::HashMap<::prost::alloc::string::String, EntityState>,
+}
+/// Ack acknowledges receipt of a StateUpdate.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Ack {
+    /// UUID of the acknowledging relay
+    #[prost(string, tag="1")]
+    pub relay_id: ::prost::alloc::string::String,
+    /// Sequence number being acknowledged
+    #[prost(uint64, tag="2")]
+    pub sequence_number: u64,
+}
+/// RetransmissionRequest requests a missed StateUpdate.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct RetransmissionRequest {
+    /// UUID of the requesting relay
+    #[prost(string, tag="1")]
+    pub relay_id: ::prost::alloc::string::String,
+    /// Sequence number of the missed update
+    #[prost(uint64, tag="2")]
+    pub sequence_number: u64,
+}
+/// Retransmission resends a StateUpdate.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Retransmission {
+    /// UUID of the sending relay
+    #[prost(string, tag="1")]
+    pub relay_id: ::prost::alloc::string::String,
+    /// The retransmitted update
+    #[prost(message, optional, tag="2")]
+    pub state_update: ::core::option::Option<StateUpdate>,
+}
+/// Heartbeat signals relay liveness.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Heartbeat {
+    /// UUID of the sending relay
+    #[prost(string, tag="1")]
+    pub relay_id: ::prost::alloc::string::String,
+    /// Time of the heartbeat
+    #[prost(message, optional, tag="2")]
+    pub timestamp: ::core::option::Option<::prost_types::Timestamp>,
+}
+/// SuspectRelay marks a relay as potentially unresponsive.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct SuspectRelay {
+    /// UUID of the suspected relay
+    #[prost(string, tag="1")]
+    pub relay_id: ::prost::alloc::string::String,
+    /// Reason for suspicion (e.g., "no heartbeat")
+    #[prost(string, tag="2")]
+    pub reason: ::prost::alloc::string::String,
+}
+/// Disconnect signals to remove a relay from the mesh.
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct Disconnect {
+    /// UUID of the relay to disconnect
+    #[prost(string, tag="1")]
+    pub relay_id: ::prost::alloc::string::String,
+    /// Reason for disconnection (e.g., "unresponsive")
+    #[prost(string, tag="2")]
+    pub reason: ::prost::alloc::string::String,
+}
+// @@protoc_insertion_point(module)

packages/server/src/proto/proto.rs

@@ -3,17 +3,17 @@
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ProtoTimestampEntry {
-    #[prost(string, tag = "1")]
+    #[prost(string, tag="1")]
     pub stage: ::prost::alloc::string::String,
-    #[prost(message, optional, tag = "2")]
+    #[prost(message, optional, tag="2")]
     pub time: ::core::option::Option<::prost_types::Timestamp>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ProtoLatencyTracker {
-    #[prost(string, tag = "1")]
+    #[prost(string, tag="1")]
     pub sequence_id: ::prost::alloc::string::String,
-    #[prost(message, repeated, tag = "2")]
+    #[prost(message, repeated, tag="2")]
     pub timestamps: ::prost::alloc::vec::Vec<ProtoTimestampEntry>,
 }
 /// MouseMove message
@@ -21,11 +21,11 @@ pub struct ProtoLatencyTracker {
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ProtoMouseMove {
     /// Fixed value "MouseMove"
-    #[prost(string, tag = "1")]
+    #[prost(string, tag="1")]
     pub r#type: ::prost::alloc::string::String,
-    #[prost(int32, tag = "2")]
+    #[prost(int32, tag="2")]
     pub x: i32,
-    #[prost(int32, tag = "3")]
+    #[prost(int32, tag="3")]
     pub y: i32,
 }
 /// MouseMoveAbs message
@@ -33,11 +33,11 @@ pub struct ProtoMouseMove {
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ProtoMouseMoveAbs {
     /// Fixed value "MouseMoveAbs"
-    #[prost(string, tag = "1")]
+    #[prost(string, tag="1")]
     pub r#type: ::prost::alloc::string::String,
-    #[prost(int32, tag = "2")]
+    #[prost(int32, tag="2")]
     pub x: i32,
-    #[prost(int32, tag = "3")]
+    #[prost(int32, tag="3")]
     pub y: i32,
 }
 /// MouseWheel message
@@ -45,11 +45,11 @@ pub struct ProtoMouseMoveAbs {
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ProtoMouseWheel {
     /// Fixed value "MouseWheel"
-    #[prost(string, tag = "1")]
+    #[prost(string, tag="1")]
     pub r#type: ::prost::alloc::string::String,
-    #[prost(int32, tag = "2")]
+    #[prost(int32, tag="2")]
     pub x: i32,
-    #[prost(int32, tag = "3")]
+    #[prost(int32, tag="3")]
     pub y: i32,
 }
 /// MouseKeyDown message
@@ -57,9 +57,9 @@ pub struct ProtoMouseWheel {
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ProtoMouseKeyDown {
     /// Fixed value "MouseKeyDown"
-    #[prost(string, tag = "1")]
+    #[prost(string, tag="1")]
     pub r#type: ::prost::alloc::string::String,
-    #[prost(int32, tag = "2")]
+    #[prost(int32, tag="2")]
     pub key: i32,
 }
 /// MouseKeyUp message
@@ -67,9 +67,9 @@ pub struct ProtoMouseKeyDown {
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ProtoMouseKeyUp {
     /// Fixed value "MouseKeyUp"
-    #[prost(string, tag = "1")]
+    #[prost(string, tag="1")]
     pub r#type: ::prost::alloc::string::String,
-    #[prost(int32, tag = "2")]
+    #[prost(int32, tag="2")]
     pub key: i32,
 }
 /// KeyDown message
@@ -77,9 +77,9 @@ pub struct ProtoMouseKeyUp {
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ProtoKeyDown {
     /// Fixed value "KeyDown"
-    #[prost(string, tag = "1")]
+    #[prost(string, tag="1")]
     pub r#type: ::prost::alloc::string::String,
-    #[prost(int32, tag = "2")]
+    #[prost(int32, tag="2")]
     pub key: i32,
 }
 /// KeyUp message
@@ -87,53 +87,53 @@ pub struct ProtoKeyDown {
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ProtoKeyUp {
     /// Fixed value "KeyUp"
-    #[prost(string, tag = "1")]
+    #[prost(string, tag="1")]
     pub r#type: ::prost::alloc::string::String,
-    #[prost(int32, tag = "2")]
+    #[prost(int32, tag="2")]
     pub key: i32,
 }
 /// Union of all Input types
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ProtoInput {
-    #[prost(oneof = "proto_input::InputType", tags = "1, 2, 3, 4, 5, 6, 7")]
+    #[prost(oneof="proto_input::InputType", tags="1, 2, 3, 4, 5, 6, 7")]
     pub input_type: ::core::option::Option<proto_input::InputType>,
 }
 /// Nested message and enum types in `ProtoInput`.
 pub mod proto_input {
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+#[derive(Clone, PartialEq, ::prost::Oneof)]
     pub enum InputType {
-        #[prost(message, tag = "1")]
+        #[prost(message, tag="1")]
         MouseMove(super::ProtoMouseMove),
-        #[prost(message, tag = "2")]
+        #[prost(message, tag="2")]
         MouseMoveAbs(super::ProtoMouseMoveAbs),
-        #[prost(message, tag = "3")]
+        #[prost(message, tag="3")]
         MouseWheel(super::ProtoMouseWheel),
-        #[prost(message, tag = "4")]
+        #[prost(message, tag="4")]
         MouseKeyDown(super::ProtoMouseKeyDown),
-        #[prost(message, tag = "5")]
+        #[prost(message, tag="5")]
         MouseKeyUp(super::ProtoMouseKeyUp),
-        #[prost(message, tag = "6")]
+        #[prost(message, tag="6")]
         KeyDown(super::ProtoKeyDown),
-        #[prost(message, tag = "7")]
+        #[prost(message, tag="7")]
         KeyUp(super::ProtoKeyUp),
     }
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ProtoMessageBase {
-    #[prost(string, tag = "1")]
+    #[prost(string, tag="1")]
     pub payload_type: ::prost::alloc::string::String,
-    #[prost(message, optional, tag = "2")]
+    #[prost(message, optional, tag="2")]
     pub latency: ::core::option::Option<ProtoLatencyTracker>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ProtoMessageInput {
-    #[prost(message, optional, tag = "1")]
+    #[prost(message, optional, tag="1")]
     pub message_base: ::core::option::Option<ProtoMessageBase>,
-    #[prost(message, optional, tag = "2")]
+    #[prost(message, optional, tag="2")]
     pub data: ::core::option::Option<ProtoInput>,
 }
 // @@protoc_insertion_point(module)