synvael/crates/server/src/client_stream.rs

// SPDX-License-Identifier: AGPL-3.0-only

//! Per-connection outbound streaming state: chunk subscription tracking and the authority-stream sink.

use std::collections::HashSet;

use net::{AuthoritySink, ChunkSink};
use shared::protocol::authority::{AuthorityMessage, ServerStats};
use shared::protocol::chunk::ChunkMessage;
use shared::world::{Chunk, ChunkData, ChunkPos};

use crate::world_server::{ServerWorld, cylinder_chunks};

/// Upper bound, in chunks, on a client's requested load radius. A larger request is clamped to this, bounding the per-client resident set and the reconcile cost the server performs on the client's behalf.
// TODO: derive from server configuration and per-tier LOD limits.
pub const SERVER_MAX_RADIUS: u16 = 24;

/// Worldgen version stamped on delivered chunk diffs. A single version exists today; this becomes the chunk's stored version once worldgen versioning lands.
const WORLDGEN_VERSION: u32 = 0;

/// Upper bound on chunks encoded and sent to one client per tick.
// TODO: replace the fixed count with a time budget once per-chunk cost varies with LOD.
const MAX_DELIVERIES_PER_TICK: usize = 32;

/// The load and drop lists produced by diffing a client's previous desired set against a new one.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct DesiredDiff {
    /// Positions newly wanted (present in the new set, absent from the previous). Delivered once resident.
    pub added: Vec<ChunkPos>,
    /// Positions no longer wanted (present in the previous set, absent from the new). The client is told to drop each it holds.
    pub removed: Vec<ChunkPos>,
}

/// Computes the load/drop diff between a client's `previous` and `new` desired sets.
#[must_use]
pub fn desired_diff<S: std::hash::BuildHasher>(
    previous: &HashSet<ChunkPos, S>,
    new: &HashSet<ChunkPos, S>,
) -> DesiredDiff {
    let mut added: Vec<ChunkPos> = new.difference(previous).copied().collect();
    let mut removed: Vec<ChunkPos> = previous.difference(new).copied().collect();
    added.sort_unstable();
    removed.sort_unstable();
    DesiredDiff { added, removed }
}

/// Tracks one connected client's chunk subscription, what has been delivered to it, and the sinks used to push to it.
pub struct ClientStream {
    /// Outbound handle onto the client's chunk stream.
    sink: ChunkSink,
    /// Outbound handle onto the client's authority stream, carried here so both per-connection sinks share one lifetime and one lookup key.
    authority: AuthoritySink,
    /// The chunk positions the client currently wants resident, already clamped to [`SERVER_MAX_RADIUS`].
    desired: HashSet<ChunkPos>,
    /// Positions already delivered to the client as [`ChunkMessage::Chunk`].
    sent: HashSet<ChunkPos>,
}

impl ClientStream {
    /// Creates a stream for a freshly connected client that has not yet subscribed.
    #[must_use]
    pub fn new(sink: ChunkSink, authority: AuthoritySink) -> Self {
        Self {
            sink,
            authority,
            desired: HashSet::new(),
            sent: HashSet::new(),
        }
    }

    /// Returns the client's current desired set, for folding into the world reconcile union.
    #[must_use]
    pub fn desired(&self) -> &HashSet<ChunkPos> {
        &self.desired
    }

    /// Applies a new subscription centered on `center` with load radius `radius`.
    ///
    /// The radius is clamped to [`SERVER_MAX_RADIUS`], the desired set is recomputed, and a [`ChunkMessage::Drop`] is emitted for every already-delivered chunk that left the set. Chunks newly entering the set are not sent here; they are delivered by [`ClientStream::flush`] once resident. Returns the number of newly-wanted positions and the number of drops emitted.
    pub fn resubscribe(&mut self, center: ChunkPos, radius: u16) -> (usize, usize) {
        let clamped = radius.min(SERVER_MAX_RADIUS);
        let mut new_desired = HashSet::new();
        cylinder_chunks(center, i32::from(clamped), &mut new_desired);

        let diff = desired_diff(&self.desired, &new_desired);
        let added = diff.added.len();
        let mut drops = 0;
        for pos in diff.removed {
            // Only chunks actually delivered need an explicit drop; positions that were wanted but never resident were never held by the client.
            if self.sent.remove(&pos) {
                self.sink.send(ChunkMessage::Drop { pos });
                drops += 1;
            }
        }
        self.desired = new_desired;
        (added, drops)
    }

    /// Delivers every desired-but-undelivered chunk that has become resident in `world`.
    ///
    /// Each chunk is encoded as a [`ChunkData`] diff against `baseline` (an all-air chunk), making the payload self-contained. Positions still pending in the worker pool are skipped and retried on a later call, as are positions beyond [`MAX_DELIVERIES_PER_TICK`]. Returns the number of chunks delivered.
    pub fn flush(&mut self, world: &ServerWorld, baseline: &Chunk) -> usize {
        // Collected first to avoid borrowing `self.desired` while mutating `self.sent`.
        let ready: Vec<ChunkPos> = self
            .desired
            .iter()
            .filter(|pos| !self.sent.contains(pos))
            .copied()
            .collect();

        let mut delivered = 0;
        for pos in ready {
            // The budget counts chunks actually encoded, so ticks where most of the desired set is still in flight are not charged for work they did not do.
            if delivered >= MAX_DELIVERIES_PER_TICK {
                break;
            }

            let Some(chunk) = world.chunk(pos) else {
                // Not resident yet; a later flush retries once the worker pool returns it.
                continue;
            };
            let data = ChunkData::from_diff(pos, WORLDGEN_VERSION, baseline, chunk);
            self.sink.send(ChunkMessage::Chunk { pos, data });
            self.sent.insert(pos);
            delivered += 1;
        }
        delivered
    }

    /// Pushes a diagnostics snapshot onto the client's authority stream.
    ///
    /// Non-blocking, and silently ignored when the connection has already gone away; see [`AuthoritySink::send`].
    pub fn send_stats(&self, stats: ServerStats) {
        self.authority.send(AuthorityMessage::ServerStats(stats));
    }
}

#[cfg(test)]
#[path = "tests/client_stream.rs"]
mod tests;