Don't drop every player after the server is held up

Extension/Multiplayer/Session/room.h

@@ -160,11 +160,17 @@ struct JoinBackoff {
         });
     }
 };
+// How much one peer may send in a second. `when` is when the packet arrived (the transport's
+// TransportMessage::arrived), not when it is read: a receiver that could not read for a few
+// seconds (a server whose console held it up, a game loading a map) reads everything that
+// arrived meanwhile at once, and counted by reading time every peer looked like a flood and
+// was dropped. Packets are not always read in arrival order (Steam's lanes are read one after
+// another): one from before the second being counted is counted in it.
 struct ReceiveBudget {
     std::uint64_t since{}, bytes{}, packets{};
-    bool accept(std::uint64_t now, std::size_t size, unsigned sources = 1) {
-        if (now - since >= 1000000) {
-            since = now;
+    bool accept(std::uint64_t when, std::size_t size, unsigned sources = 1) {
+        if (when > since && when - since >= 1000000) {
+            since = when;
             bytes = packets = 0;
         }
         bytes += size;

Extension/Multiplayer/Session/session_receive.cpp

@@ -334,7 +334,7 @@ void networking(Session &s, const NativeFrame &local, std::uint64_t now) {
         if (!link)
             continue;
         const bool direct_link = s.mode == Mode::join && message.peer != s.host_id;
-        if (!link->budget.accept(now, message.bytes.size(),
+        if (!link->budget.accept(message.arrived ? message.arrived : now, message.bytes.size(),
                                  s.mode == Mode::join && !direct_link ? max_remote_players : 1U)) {
             disconnect(s, message.peer, "Peer exceeded the multiplayer packet limit.");
             if (s.mode == Mode::off)

Extension/Multiplayer/Steam/steam_transport.cpp

@@ -696,7 +696,8 @@ std::vector<TransportMessage> SteamTransport::receive() {
                 continue;
             }
             const auto *data = static_cast<const std::uint8_t *>(message->m_pData);
-            result.push_back({id, {data, data + message->m_cbSize}});
+            result.push_back({id, {data, data + message->m_cbSize},
+                              message->m_usecTimeReceived > 0 ? static_cast<std::uint64_t>(message->m_usecTimeReceived) : 0});
             ++p.state.received;
             p.state.received_bytes += static_cast<std::uint64_t>(message->m_cbSize);
         }

Extension/Multiplayer/Steam/steam_transport.h

@@ -16,6 +16,10 @@ struct TransportPeer {
 struct TransportMessage {
     std::uint64_t peer{};
     std::vector<std::uint8_t> bytes;
+    // When Steam received it, in microseconds on a clock of its own (only differences mean
+    // anything); 0 when the transport does not say. A receiver that was kept from reading for a
+    // while gets everything that arrived meanwhile in one go: this is what tells that from a flood.
+    std::uint64_t arrived{};
 };
 // One message of SteamTransport::send_batch, with send()'s arguments; `sent` receives what
 // send() would have returned for it.

Extension/Multiplayer/Test/mesh_tests.cpp

@@ -89,6 +89,8 @@ struct SimulatedNetwork {
     static inline std::uint64_t hold_control_to{};
     static inline std::array<unsigned, 3> lane_sends{};
     static inline std::map<std::uint64_t, std::uint64_t> queues;
+    // What a message's arrival time is (TransportMessage::arrived); 0: the transport does not say.
+    static inline std::uint64_t clock{};
     static std::pair<std::uint64_t, std::uint64_t> pair(std::uint64_t a, std::uint64_t b) {
         return {std::min(a, b), std::max(a, b)};
     }
@@ -200,7 +202,7 @@ bool SteamTransport::send(std::uint64_t id, std::span<const std::uint8_t> bytes,
         return true;
     if (!reliable && SimulatedNetwork::lose_unreliable && ++p.unreliable % 5 == 0)
         return true;
-    Impl::bus.at(id)->inbox.push_back({{p.state.local_id, {bytes.begin(), bytes.end()}}, lane});
+    Impl::bus.at(id)->inbox.push_back({{p.state.local_id, {bytes.begin(), bytes.end()}, SimulatedNetwork::clock}, lane});
     return true;
 }
 void SteamTransport::send_batch(std::span<TransportSend> messages) {
@@ -469,6 +471,8 @@ struct Simulation {
     NativeFrame local;
     std::vector<float> positions;
     unsigned capacity, tps;
+    std::size_t held = static_cast<std::size_t>(-1); // a node that is not run: held up, reading nothing
+    bool stamp{};                                     // messages carry when they arrived
     explicit Simulation(unsigned limit = max_players, unsigned rate = 20) : capacity(limit), tps(rate) {
         local.ready = true;
         local.pose.skater.resize(395);
@@ -503,9 +507,10 @@ struct Simulation {
     void run(unsigned frames) {
         for (unsigned i = 0; i < frames; ++i) {
             now += network_tick_us;
+            SimulatedNetwork::clock = stamp ? now : 0;
             for (std::size_t n = 0; n < nodes.size(); ++n) {
                 auto &s = *nodes[n];
-                if (s.mode == Mode::off)
+                if (s.mode == Mode::off || n == held)
                     continue;
                 local.pose.root.position[0] = static_cast<float>(now - 10000000) / 1000000;
                 local.pose.root.position[2] = n < positions.size() ? positions[n] : static_cast<float>(n);
@@ -639,6 +644,56 @@ void party_checks() {
     check(!host.local_party && host.parties.parties().empty(), "A guest who left the lobby stayed in the host's party");
     std::cout << "Lobby parties: nobody by default, invites, two parties, party chat, leaving and departures passed.\n";
 }
+// A host that is kept from reading for a few seconds (a server whose console held it up) then
+// reads everything its guests sent meanwhile in one go. That is not a flood: nobody is dropped
+// for it, while a real flood still is.
+void stall_checks() {
+    {
+        // 16 s of a player's ordinary traffic read at once, counted by when it arrived.
+        ReceiveBudget backlog;
+        bool kept = true;
+        for (std::uint64_t i = 0; i < 16 * 90; ++i) kept = kept && backlog.accept(5000000 + i * (1000000 / 90), 300);
+        check(kept, "A backlog read after a stall was taken for a flood");
+        ReceiveBudget flood;
+        bool refused = false;
+        for (std::uint64_t i = 0; i < 2000; ++i) refused = refused || !flood.accept(5000000 + i * 100, 300);
+        check(refused, "A flood was let through the packet limit");
+        // Lanes are read one after another, so arrival times step back and forth.
+        ReceiveBudget lanes;
+        refused = false;
+        for (std::uint64_t i = 0; i < 2000; ++i) refused = refused || !lanes.accept(i % 2 ? 9000000 : 9500000, 300);
+        check(refused, "Packets read out of arrival order started the count again");
+    }
+    Simulation sim(max_players, 120);
+    sim.stamp = true;
+    for (unsigned i = 0; i < 4; ++i) sim.add();
+    sim.run(240);
+    auto &host = *sim.nodes[0];
+    const auto connected = [&] {
+        unsigned count{};
+        for (std::size_t n = 1; n < sim.nodes.size(); ++n) {
+            const auto *peer = find_peer(host, sim.nodes[n]->transport.status().local_id);
+            count += peer && peer->handshaken && sim.nodes[n]->mode == Mode::join;
+        }
+        return count;
+    };
+    check(connected() == 3, "The stall fixture did not connect its guests");
+    // Nine seconds unread: under the ten after which a silent player is given up on, and more
+    // than a second's allowance from every guest.
+    const auto sent = sim.nodes[1]->transport.status().sent;
+    sim.held = 0;
+    sim.run(static_cast<unsigned>(9000000 / network_tick_us));
+    sim.held = static_cast<std::size_t>(-1);
+    check(sim.nodes[1]->transport.status().sent - sent > 2ULL * multiplayer_tick_rates.back() + 80 + 32,
+          "The stall fixture did not queue more than the packet limit");
+    sim.run(240);
+    if (host.mode != Mode::host || connected() != 3)
+        for (std::size_t n = 0; n < sim.nodes.size(); ++n)
+            std::cerr << "  node " << n << " mode " << static_cast<int>(sim.nodes[n]->mode) << ": " << sim.nodes[n]->status << "\n";
+    check(host.mode == Mode::host && connected() == 3, "Guests were dropped after their host was held up for a few seconds");
+    SimulatedNetwork::clock = 0;
+    std::cout << "Stalls: a held-up host keeps its guests; floods and out-of-order arrival are still limited.\n";
+}
 // What a host changes outside its physics tuning (the trainer's class values and trick
 // multipliers) reaches its guests while it sets everyone's physics: when it changes, to a
 // player who joins later, and from nobody but the host.
@@ -1934,6 +1989,7 @@ int main(int argc, char **argv) {
         dingosdk::multiplayer::throwdown_routing_checks();
         dingosdk::multiplayer::party_checks();
         dingosdk::multiplayer::physics_extras_checks();
+        dingosdk::multiplayer::stall_checks();
         dingosdk::multiplayer::scoring_checks();
         dingosdk::multiplayer::object_sync_checks();
         dingosdk::multiplayer::session_controls_checks();

Server/main.cpp

@@ -23,6 +23,7 @@
 #endif
 #include <atomic>
 #include <chrono>
+#include <condition_variable>
 #include <cstdio>
 #include <ctime>
 #include <deque>
@@ -44,6 +45,57 @@ constexpr int restart_for_update = -2;
 std::string update_version;
 std::mutex log_mutex;
 std::ofstream log_file;
+// The console is written by a thread of its own. Windows holds a console's output while text
+// in its window is selected, and a pipe nobody reads fills up: whoever writes then waits. When
+// that was the server loop, the server read nothing from its players until the console let go,
+// and they were gone by then. The log file is still written where the line is made.
+struct Console {
+    std::mutex mutex;
+    std::condition_variable more, drained;
+    std::deque<std::string> lines;
+    std::size_t skipped{};
+    bool writing{}, started{};
+    void write(std::string line) {
+        std::lock_guard lock(mutex);
+        if (!started) {
+            started = true;
+            std::thread([this] { run(); }).detach();
+        }
+        // A console that stays held must not grow this without end: the oldest lines go, and
+        // the file has them all.
+        if (lines.size() >= 4096) {
+            lines.pop_front();
+            ++skipped;
+        }
+        lines.push_back(std::move(line));
+        more.notify_one();
+    }
+    void run() {
+        for (;;) {
+            std::unique_lock lock(mutex);
+            more.wait(lock, [&] { return !lines.empty(); });
+            const auto line = std::move(lines.front());
+            lines.pop_front();
+            const auto missed = std::exchange(skipped, 0);
+            writing = true;
+            lock.unlock();
+            if (missed) std::printf("(%zu earlier lines are only in ReSkateServer.log: the console was not taking output)\n", missed);
+            std::fputs(line.c_str(), stdout);
+            lock.lock();
+            writing = false;
+            if (lines.empty()) drained.notify_all();
+        }
+    }
+    // Waits until everything written so far is on screen, but not for a console that is held.
+    void flush(std::chrono::milliseconds limit) {
+        std::unique_lock lock(mutex);
+        drained.wait_for(lock, limit, [&] { return lines.empty() && !writing; });
+    }
+};
+Console &console() {
+    static auto *value = new Console; // outlives every thread that may still be writing at exit
+    return *value;
+}
 void write_log(const std::string &text) {
     std::lock_guard lock(log_mutex);
     const auto now = std::time(nullptr);
@@ -55,8 +107,8 @@ void write_log(const std::string &text) {
 #endif
     char stamp[32]{};
     std::strftime(stamp, sizeof stamp, "%Y-%m-%d %H:%M:%S", &local);
-    std::printf("[%s] %s\n", stamp + 11, text.c_str());
     if (log_file) log_file << '[' << stamp << "] " << text << std::endl;
+    console().write(std::string("[") + (stamp + 11) + "] " + text + "\n");
 }
 std::atomic<bool> finished{};
 #ifdef _WIN32
@@ -444,6 +496,7 @@ int wmain(int argc, wchar_t **argv) {
             write_log("Installing server update " + update_version + "...");
             install_update(folder());
             write_log("Server update " + update_version + " installed; starting it.");
+            console().flush(std::chrono::seconds(2));
             if (relaunch()) {
                 code = 0;
                 break;
@@ -462,9 +515,10 @@ int wmain(int argc, wchar_t **argv) {
     if (code != 0 && !stopping && own_window()) {
         auto &input = console_input();
         input.take();
-        std::printf("Press Enter to close.\n");
+        console().write("Press Enter to close.\n");
         while (!stopping && input.take().empty()) Sleep(50);
     }
+    console().flush(std::chrono::seconds(2));
     return code;
 }
 #else
@@ -478,6 +532,7 @@ int main(int argc, char **argv) {
             write_log("Installing server update " + update_version + "...");
             install_update(folder());
             write_log("Server update " + update_version + " installed; starting it.");
+            console().flush(std::chrono::seconds(2));
             if (relaunch()) {
                 code = 0;
                 break;
@@ -491,6 +546,7 @@ int main(int argc, char **argv) {
         }
     }
     finished = true;
+    console().flush(std::chrono::seconds(2));
     return code;
 }
 #endif

Server/server_host.cpp

@@ -414,10 +414,12 @@ bool Host::accept_data(Guest &source, const Packet &p) {
     }
     return true;
 }
-void Host::receive(std::uint64_t peer, std::span<const std::uint8_t> bytes) {
+void Host::receive(std::uint64_t peer, std::span<const std::uint8_t> bytes, std::uint64_t received_at) {
     auto *link = find(peer);
     if (!link) return;
-    if (!link->budget.accept(now_, bytes.size(), 1U)) return drop(peer, "Peer exceeded the multiplayer packet limit.");
+    // Counted by arrival: after the server was held up, everyone's packets are read at once.
+    if (!link->budget.accept(received_at ? received_at : now_, bytes.size(), 1U))
+        return drop(peer, "Peer exceeded the multiplayer packet limit.");
     bool missing_reference{};
     const auto decoded = link->receiver.receive(bytes, missing_reference, world_);
     if (missing_reference) return;
@@ -764,7 +766,7 @@ void Host::tick(std::uint64_t now) {
         }
         if (link.connected && !guest->connected_at) guest->connected_at = now_;
     }
-    for (const auto &message : transport_.receive()) receive(message.peer, message.bytes);
+    for (const auto &message : transport_.receive()) receive(message.peer, message.bytes, message.arrived);
     receive_cosmetics();
     for (auto it = guests_.begin(); it != guests_.end();) {
         auto &g = *(it++)->second;

Server/server_host.h

@@ -158,7 +158,8 @@ class Host {
     void change_map(std::string_view map); // a level name, level path or destination
     bool same_map(std::string_view asset) const { return map_hash(map_destination(asset)) == map_; }
     bool accept_data(Guest &source, const Packet &);
-    void receive(std::uint64_t peer, std::span<const std::uint8_t> bytes);
+    // `received_at`: the transport's arrival time for the message (TransportMessage::arrived).
+    void receive(std::uint64_t peer, std::span<const std::uint8_t> bytes, std::uint64_t received_at);
     void receive_cosmetics();
     void sync_objects();
     void apply_layers();