feat: count how a peer compressed its first batches

CMakeLists.txt

@@ -93,6 +93,9 @@ if (FALCON_NETWORK_BUILD_TESTS)
     add_executable(FalconSharedPacketTest tests/SharedPacket.cpp)
     target_link_libraries(FalconSharedPacketTest PRIVATE FalconNetwork)
     add_test(NAME shared-packet COMMAND FalconSharedPacketTest)
+    add_executable(FalconBatchTraitsTest tests/BatchTraits.cpp)
+    target_link_libraries(FalconBatchTraitsTest PRIVATE FalconNetwork)
+    add_test(NAME batch-traits COMMAND FalconBatchTraitsTest)
     add_executable(FalconSignalingDeadlineTest tests/SignalingDeadline.cpp)
     target_link_libraries(FalconSignalingDeadlineTest PRIVATE FalconNetwork)
     add_test(NAME signaling-deadline COMMAND FalconSignalingDeadlineTest)

include/Network/CompressedNetworkPeer.h

@@ -14,6 +14,18 @@ class CompressedNetworkPeer : public NetworkPeer {
 
     static const unsigned char GAME_PACKET_ID = 0xfe;
 
+    // how the other side compressed the batches it sent, counted over its first batches
+    struct BatchTraits {
+        unsigned short mBatches = 0;
+        // compressed although shorter than the threshold it was given
+        unsigned short mCompressedBelowThreshold = 0;
+        // an algorithm other than the one it was told to use
+        unsigned short mOtherAlgorithm = 0;
+        // deflate batches compared with what the game's compressor writes for the same bytes
+        unsigned short mCompared = 0;
+        unsigned short mNotReproduced = 0;
+    };
+
     explicit CompressedNetworkPeer(std::shared_ptr<NetworkPeer> peer);
 
     void enableCompression(CompressionAlgorithm algorithm, unsigned short threshold);
@@ -22,6 +34,8 @@ class CompressedNetworkPeer : public NetworkPeer {
 
     bool isCompressionEnabled() const { return mCompressionEnabled; }
 
+    const BatchTraits &getBatchTraits() const { return mBatchTraits; }
+
     void sendPacket(const std::string &data, Reliability reliability, Compressibility compressibility) override;
 
     DataStatus receivePacket(std::string &outData) override;
@@ -43,10 +57,13 @@ class CompressedNetworkPeer : public NetworkPeer {
 
     static bool _compressSnappy(const std::string &input, std::string &outData);
 
+    void _observe(CompressionAlgorithm algorithm, const char *body, size_t length, const std::string &batch);
+
     static bool _decompressSnappy(const char *data, size_t length, std::string &outData);
 
     std::shared_ptr<NetworkPeer> mPeer;
     CompressionAlgorithm mAlgorithm;
     unsigned short mCompressionThreshold;
     bool mCompressionEnabled;
+    BatchTraits mBatchTraits;
 };

src/Network/CompressedNetworkPeer.cpp

@@ -5,11 +5,67 @@
 #include <snappy.h>
 #include <zlib.h>
 
+#include <cstdint>
+
 namespace {
 
     const int RAW_DEFLATE_WINDOW_BITS = -15;
     const size_t MAX_DECOMPRESSED_SIZE = 8 * 1024 * 1024;
 
+    // sub_1404C8960: the game deflates a batch at level 7, raw, memory level 8, in one go
+    const int GAME_DEFLATE_LEVEL = 7;
+    const int GAME_MEMORY_LEVEL = 8;
+    // the first batches say enough, and a large one costs too much to deflate again
+    const unsigned short OBSERVED_BATCHES = 64;
+    const size_t MAX_COMPARED_SIZE = 256 * 1024;
+
+    bool deflateLikeGame(const char *data, size_t length, std::string &outData) {
+        z_stream stream{};
+        if (deflateInit2(&stream, GAME_DEFLATE_LEVEL, Z_DEFLATED, RAW_DEFLATE_WINDOW_BITS, GAME_MEMORY_LEVEL,
+                         Z_DEFAULT_STRATEGY)
+            != Z_OK)
+            return false;
+
+        outData.resize(deflateBound(&stream, (uLong) length));
+        stream.next_in = (Bytef *) data;
+        stream.avail_in = (uInt) length;
+        stream.next_out = (Bytef *) &outData[0];
+        stream.avail_out = (uInt) outData.size();
+        const int result = deflate(&stream, Z_FINISH);
+        const size_t written = outData.size() - stream.avail_out;
+        deflateEnd(&stream);
+        if (result != Z_STREAM_END)
+            return false;
+        outData.resize(written);
+        return true;
+    }
+
+    // a zlib that is not the reference one (a drop-in replacement) writes other bytes for the same input:
+    // comparing against it would say nothing
+    bool deflatesLikeReference() {
+        static const bool same = []() {
+            std::string input(20000, '\0');
+            static const char WORD[] = "minecraft:stone_bricks";
+            uint32_t state = 0x12345678;
+            for (size_t index = 0; index < input.size(); index++) {
+                state = state * 1664525u + 1013904223u;
+                input[index] = index % 97 < 60 ? WORD[index % 22] : (char) (state >> 24);
+            }
+
+            std::string output;
+            if (!deflateLikeGame(input.data(), input.size(), output) || output.size() != 8269)
+                return false;
+            uint32_t hash = 2166136261u;
+            for (const char byte: output) {
+                hash ^= (unsigned char) byte;
+                hash *= 16777619u;
+            }
+            // zlib 1.3.1, the version the game is built with
+            return hash == 0x55829a2c;
+        }();
+        return same;
+    }
+
 }
 
 CompressedNetworkPeer::CompressedNetworkPeer(std::shared_ptr<NetworkPeer> peer)
@@ -200,19 +256,43 @@ NetworkPeer::DataStatus CompressedNetworkPeer::receivePacket(std::string &outDat
         if (algorithm == CompressionAlgorithm::ZLib) {
             if (!_decompressZlib(body, bodyLength, outData))
                 continue;
+            _observe(algorithm, body, bodyLength, outData);
             return DataStatus::HasData;
         }
 
         if (algorithm == CompressionAlgorithm::Snappy) {
             if (!_decompressSnappy(body, bodyLength, outData))
                 continue;
+            _observe(algorithm, body, bodyLength, outData);
             return DataStatus::HasData;
         }
 
         continue;
     }
 }
 
+void CompressedNetworkPeer::_observe(CompressionAlgorithm algorithm, const char *body, size_t length,
+                                     const std::string &batch) {
+    if (mBatchTraits.mBatches >= OBSERVED_BATCHES)
+        return;
+    mBatchTraits.mBatches++;
+
+    // sub_1404C8960: a batch shorter than the threshold goes out as it is
+    if (batch.size() < mCompressionThreshold)
+        mBatchTraits.mCompressedBelowThreshold++;
+    if (algorithm != mAlgorithm)
+        mBatchTraits.mOtherAlgorithm++;
+
+    if (algorithm != CompressionAlgorithm::ZLib || batch.size() > MAX_COMPARED_SIZE || !deflatesLikeReference())
+        return;
+    std::string expected;
+    if (!deflateLikeGame(batch.data(), batch.size(), expected))
+        return;
+    mBatchTraits.mCompared++;
+    if (expected.size() != length || expected.compare(0, length, body, length) != 0)
+        mBatchTraits.mNotReproduced++;
+}
+
 NetworkPeer::NetworkStatus CompressedNetworkPeer::getNetworkStatus() const {
     return mPeer->getNetworkStatus();
 }

tests/BatchTraits.cpp

@@ -0,0 +1,123 @@
+#include "Network/CompressedNetworkPeer.h"
+
+#include <zlib.h>
+
+#include <cstdint>
+#include <deque>
+#include <iostream>
+#include <memory>
+#include <stdexcept>
+#include <string>
+
+namespace {
+    void require(bool condition, const char *message) {
+        if (!condition)
+            throw std::runtime_error(message);
+    }
+
+    class WirePeer : public NetworkPeer {
+    public:
+        void sendPacket(const std::string &data, Reliability, Compressibility) override {
+            mIncoming.push_back(data);
+        }
+
+        DataStatus receivePacket(std::string &outData) override {
+            if (mIncoming.empty())
+                return DataStatus::NoData;
+            outData = mIncoming.front();
+            mIncoming.pop_front();
+            return DataStatus::HasData;
+        }
+
+        NetworkStatus getNetworkStatus() const override {
+            return NetworkStatus();
+        }
+
+        bool usesGamePacketId() const override {
+            return false;
+        }
+
+        std::deque<std::string> mIncoming;
+    };
+
+    // text with noise in it: levels differ on it, on a run of one byte they would not
+    std::string batchOf(size_t size) {
+        std::string batch(size, '\0');
+        static const char WORD[] = "minecraft:stone_bricks";
+        uint32_t state = 7;
+        for (size_t index = 0; index < size; index++) {
+            state = state * 1664525u + 1013904223u;
+            batch[index] = index % 97 < 60 ? WORD[index % 22] : (char) (state >> 24);
+        }
+        return batch;
+    }
+
+    std::string deflated(const std::string &input, int level) {
+        z_stream stream{};
+        require(deflateInit2(&stream, level, Z_DEFLATED, -15, 8, Z_DEFAULT_STRATEGY) == Z_OK, "deflate did not start");
+        std::string out(deflateBound(&stream, (uLong) input.size()), '\0');
+        stream.next_in = (Bytef *) input.data();
+        stream.avail_in = (uInt) input.size();
+        stream.next_out = (Bytef *) &out[0];
+        stream.avail_out = (uInt) out.size();
+        require(deflate(&stream, Z_FINISH) == Z_STREAM_END, "deflate did not finish");
+        out.resize(out.size() - stream.avail_out);
+        deflateEnd(&stream);
+        return out;
+    }
+
+    std::string receive(CompressedNetworkPeer &peer) {
+        std::string batch;
+        require(peer.receivePacket(batch) == NetworkPeer::DataStatus::HasData, "a batch was lost");
+        return batch;
+    }
+}
+
+int main() {
+    try {
+        const std::shared_ptr<WirePeer> wire = std::make_shared<WirePeer>();
+        CompressedNetworkPeer peer(wire);
+        peer.enableCompression(CompressedNetworkPeer::CompressionAlgorithm::ZLib, 256);
+
+        const std::string large = batchOf(6000);
+        const std::string small = batchOf(100);
+        // a neighbouring level often writes the same bytes: a far one does not
+        require(deflated(large, 7) != deflated(large, 1), "the two levels write the same bytes");
+
+        // as the game writes it
+        wire->mIncoming.push_back(std::string(1, '\0') + deflated(large, 7));
+        require(receive(peer) == large, "the batch changed");
+        require(peer.getBatchTraits().mBatches == 1 && peer.getBatchTraits().mCompared == 1, "the batch was not compared");
+        require(peer.getBatchTraits().mNotReproduced == 0, "the game's own bytes were not reproduced");
+        require(peer.getBatchTraits().mCompressedBelowThreshold == 0, "a long batch counted as short");
+
+        // another level
+        wire->mIncoming.push_back(std::string(1, '\0') + deflated(large, 1));
+        require(receive(peer) == large, "the batch changed");
+        require(peer.getBatchTraits().mNotReproduced == 1, "another level was taken for the game's");
+
+        // a short batch the game would send as it is
+        wire->mIncoming.push_back(std::string(1, '\0') + deflated(small, 7));
+        require(receive(peer) == small, "the batch changed");
+        require(peer.getBatchTraits().mCompressedBelowThreshold == 1, "a short compressed batch went unseen");
+
+        // sent as it is: nothing to count
+        wire->mIncoming.push_back(std::string(1, '\xff') + small);
+        require(receive(peer) == small, "the batch changed");
+        require(peer.getBatchTraits().mBatches == 3, "a batch sent as it is was counted");
+
+        // the other algorithm
+        std::string snappy;
+        require(CompressedNetworkPeer::compress(CompressedNetworkPeer::CompressionAlgorithm::Snappy, large, snappy),
+                "snappy did not compress");
+        wire->mIncoming.push_back(std::string(1, '\x01') + snappy);
+        require(receive(peer) == large, "the batch changed");
+        require(peer.getBatchTraits().mOtherAlgorithm == 1, "the other algorithm went unseen");
+    } catch (const std::exception &error) {
+        std::cout << "FAIL: " << error.what() << std::endl;
+        return 1;
+    }
+
+    std::cout << "PASS: batch traits" << std::endl;
+    return 0;
+}