okooooooo/EVEngine
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
1a97146aa3ba6ce89c7fa3f196558ba43ce034ff
EVEngine/test/pixelworld_streaming.cpp
T

308 lines
16 KiB
C++
Raw Blame History

#include "zeroerr/assert.h"
#include "zeroerr/unittest.h"
#include "pixelworld/PixelWorld.h"
#include "pixelworld_streaming/PixelWorldStreaming.h"
#include "pixelworld_streaming/PixelWorldTransport.h"
#include <algorithm>
#include <cstdint>
#include <string_view>
using namespace eve::pixelworld;
using namespace eve::pixelworld_streaming;
namespace {
eve::SnapshotHashProvider archiveHashProvider() {
return [](std::string_view input) -> eve::Result<eve::ContentId> {
std::uint64_t left = 14695981039346656037ULL;
std::uint64_t right = 1099511628211ULL;
for (const unsigned char byte : input) {
left = (left ^ byte) * 1099511628211ULL;
right = (right ^ (std::uint64_t(byte) + 0x9e3779b97f4a7c15ULL)) * 14029467366897019727ULL;
}
eve::ContentId::Bytes bytes{};
for (int index = 0; index < 8; ++index) {
bytes[std::size_t(index)] = std::uint8_t(left >> (56 - index * 8));
bytes[std::size_t(index + 8)] = std::uint8_t(right >> (56 - index * 8));
}
return eve::Result<eve::ContentId>::success(eve::ContentId(bytes));
};
}
eve::LogicalId chunkArchiveSchema() {
return *eve::LogicalId::parse("pixelworld:chunk-batch");
}
} // namespace
TEST_CASE("pixelworld_streaming_interest_move_emits_entry_and_eviction_correction") {
PixelWorld authority(8301);
authority.setMaterial(1, 1, "stone");
authority.setMaterial(2 * kPixelChunkSize + 1, 1, "water");
PixelChunkStreamCursor cursor;
auto first = cursor.capture(authority, {0, 0, 0, 0}).expect("capture first interest");
CHECK(first.fullResync);
REQUIRE_EQ(first.batch.chunks.size(), std::size_t(1));
PixelWorld replica(8301);
replica.applyChunkBatch(first.batch, 0).expect("apply first interest");
CHECK_EQ(replica.getMaterial(1, 1), int(MaterialId::Stone));
CHECK_EQ(replica.getMaterial(2 * kPixelChunkSize + 1, 1), int(MaterialId::Air));
auto moved = cursor.capture(authority, {2, 0, 2, 0}).expect("move interest");
CHECK(!moved.fullResync);
CHECK_EQ(moved.chunksEntered, std::uint32_t(1));
CHECK_EQ(moved.chunksEvicted, std::uint32_t(1));
REQUIRE_EQ(moved.batch.chunks.size(), std::size_t(2));
replica.applyChunkBatch(moved.batch, replica.revision()).expect("apply moved interest");
CHECK_EQ(replica.getMaterial(1, 1), int(MaterialId::Air));
CHECK_EQ(replica.getMaterial(2 * kPixelChunkSize + 1, 1), int(MaterialId::Water));
const auto unchanged = cursor.capture(authority, {2, 0, 2, 0}).expect("unchanged interest");
CHECK(unchanged.batch.chunks.empty());
}
TEST_CASE("pixelworld_streaming_disconnect_reconnect_applies_authoritative_chunk_delta") {
PixelWorld authority(8302);
authority.setMaterial(4, 4, "wood");
PixelChunkStreamCursor cursor;
auto initial = cursor.capture(authority, {0, 0, 0, 0}).expect("initial sync");
PixelWorld replica(8302);
replica.applyChunkBatch(initial.batch, 0).expect("apply initial sync");
authority.setMaterial(4, 4, "fire");
authority.setMaterial(5, 4, "oil");
auto reconnect = cursor.capture(authority, {0, 0, 0, 0}).expect("reconnect delta");
REQUIRE_EQ(reconnect.batch.chunks.size(), std::size_t(1));
replica.applyChunkBatch(reconnect.batch, replica.revision()).expect("apply reconnect correction");
CHECK_EQ(replica.getCell(4, 4), authority.getCell(4, 4));
CHECK_EQ(replica.getCell(5, 4), authority.getCell(5, 4));
CHECK_EQ(replica.revision(), authority.revision());
}
TEST_CASE("pixelworld_streaming_epoch_change_forces_full_resync_and_stale_writer_rejects") {
PixelWorld authority(8303);
authority.setMaterial(3, 3, "sand");
const auto saved = authority.saveSnapshot().expect("save authority");
PixelChunkStreamCursor cursor;
auto initial = cursor.capture(authority, {0, 0, 0, 0}).expect("initial");
CHECK(initial.fullResync);
authority.restoreSnapshot(saved).expect("restore changes epoch");
auto rebuilt = cursor.capture(authority, {0, 0, 0, 0}).expect("full rebuild");
CHECK(rebuilt.fullResync);
REQUIRE_EQ(rebuilt.batch.chunks.size(), std::size_t(1));
PixelWorld replica(8303);
replica.setMaterial(10, 10, "stone");
const auto before = replica.saveSnapshot().expect("before stale correction");
CHECK(!replica.applyChunkBatch(rebuilt.batch, 0).ok());
const auto after = replica.saveSnapshot().expect("after stale correction");
REQUIRE_EQ(before.size(), after.size());
CHECK(std::equal(before.begin(), before.end(), after.begin()));
}
TEST_CASE("pixelworld_streaming_full_resync_atomically_clears_projection_and_accepts_rewind") {
PixelWorld authority(8304);
authority.setMaterial(3, 3, "stone");
PixelChunkStreamCursor cursor;
const auto populated = cursor.capture(authority, {0, 0, 0, 0}).expect("populated sync");
REQUIRE(populated.batch.fullResync);
PixelWorld replica(8304);
replica.applyChunkBatch(populated.batch, 0).expect("apply populated projection");
CHECK_EQ(replica.getMaterial(3, 3), int(MaterialId::Stone));
PixelWorld emptyAuthority(8304);
const auto emptySnapshot = emptyAuthority.saveSnapshot().expect("empty authority snapshot");
authority.restoreSnapshot(emptySnapshot).expect("rewind authority to empty epoch");
const auto reset = cursor.capture(authority, {0, 0, 0, 0}).expect("rewound full sync");
CHECK(reset.fullResync);
CHECK(reset.batch.fullResync);
CHECK_EQ(reset.batch.sourceRevision, std::uint64_t(0));
CHECK(reset.batch.chunks.empty());
replica.applyChunkBatch(reset.batch, replica.revision()).expect("replace projection after rewind");
CHECK_EQ(replica.revision(), std::uint64_t(0));
CHECK_EQ(replica.chunkCount(), std::size_t(0));
CHECK_EQ(replica.getMaterial(3, 3), int(MaterialId::Air));
}
TEST_CASE("pixelworld_streaming_zstd_chunk_archive_round_trips_and_is_smaller_than_raw") {
const auto hash = archiveHashProvider();
PixelWorld authority(8401);
for (int y = 0; y < kPixelChunkSize; ++y)
for (int x = 0; x < kPixelChunkSize; ++x)
authority.setCell(x, y, {MaterialId::Stone, 20, 0});
authority.setCell(17, 23, {MaterialId::Stone, 20, 0, 73});
PixelChunkStreamCursor cursor;
const auto update = cursor.capture(authority, {0, 0, 0, 0}).expect("capture archive batch");
const auto compressed = archiveChunkBatch(update.batch, eve::PersistentId::nil(),
PixelChunkArchiveCodec::Zstd, hash)
.expect("compress archive");
const auto raw = archiveChunkBatch(update.batch, eve::PersistentId::nil(),
PixelChunkArchiveCodec::None, hash)
.expect("raw archive");
const auto compressedJson = eve::serializeSnapshotEnvelope(compressed).expect("serialize compressed");
const auto rawJson = eve::serializeSnapshotEnvelope(raw).expect("serialize raw");
CHECK(compressedJson.size() < rawJson.size() / 4);
const auto parsed = eve::parseSnapshotEnvelope(compressedJson, hash).expect("parse compressed");
const auto decoded = decodeChunkBatchArchive(parsed, hash).expect("decode archive");
PixelWorld replica(8401);
replica.applyChunkBatch(decoded, 0).expect("apply decoded archive");
CHECK_EQ(replica.getCell(17, 23), authority.getCell(17, 23));
CHECK_EQ(replica.getCell(17, 23).thermalRemainder, std::uint16_t(73));
CHECK_EQ(replica.revision(), authority.revision());
}
TEST_CASE("pixelworld_streaming_chunk_archive_v1_migrates_and_corruption_is_isolated") {
const auto hash = archiveHashProvider();
PixelWorld authority(8402);
authority.setMaterial(3, 4, "water");
PixelChunkStreamCursor cursor;
const auto update = cursor.capture(authority, {0, 0, 0, 0}).expect("capture");
const auto current = archiveChunkBatch(update.batch, eve::PersistentId::nil(),
PixelChunkArchiveCodec::Zstd, hash)
.expect("archive");
auto legacyPayload = current.payload;
legacyPayload.getIf<eve::Value::Object>()->erase("sourceLastEditSequence");
legacyPayload.getIf<eve::Value::Object>()->erase("fullResync");
const auto legacy = eve::makeSnapshotEnvelope(
"pixelworld.chunk-batch", chunkArchiveSchema(), eve::SchemaVersion(1),
eve::PersistentId::nil(), current.revision, current.tick,
std::move(legacyPayload), hash)
.expect("seal legacy");
const auto migrated = decodeChunkBatchArchive(legacy, hash).expect("decode legacy");
CHECK_EQ(migrated.sourceLastEditSequence, std::uint64_t(0));
REQUIRE_EQ(migrated.chunks.size(), std::size_t(1));
auto corruptPayload = current.payload;
auto& chunks = *corruptPayload.getIf<eve::Value::Object>()->at("chunks").getIf<eve::Value::Array>();
auto& stored = *chunks.front().getIf<eve::Value::Object>()->at("stored").getIf<std::string>();
REQUIRE(!stored.empty());
stored[0] = stored[0] == '0' ? '1' : '0';
const auto resealedCorrupt = eve::makeSnapshotEnvelope(
"pixelworld.chunk-batch", chunkArchiveSchema(), eve::SchemaVersion(2),
eve::PersistentId::nil(), current.revision, current.tick,
std::move(corruptPayload), hash)
.expect("seal corrupt physical payload");
PixelWorld untouched(8402);
untouched.setMaterial(9, 9, "wood");
const auto before = untouched.saveSnapshot().expect("before corrupt decode");
CHECK(!decodeChunkBatchArchive(resealedCorrupt, hash).ok());
const auto after = untouched.saveSnapshot().expect("after corrupt decode");
CHECK(std::equal(before.begin(), before.end(), after.begin()));
auto badOuterHash = current;
badOuterHash.contentHash = eve::ContentId::nil();
CHECK(!decodeChunkBatchArchive(badOuterHash, hash).ok());
}
TEST_CASE("pixelworld_transport_retransmits_loss_and_commits_complete_transfer_atomically") {
PixelWorld authority(8501);
authority.setMaterial(1, 1, "stone");
authority.setMaterial(kPixelChunkSize + 1, 1, "water");
authority.setMaterial(2 * kPixelChunkSize + 1, 1, "wood");
auto sender = ReliablePixelChunkSender::create(91).expect("create sender");
auto receiver = ReliablePixelChunkReceiver::create(91).expect("create receiver");
const auto parts = sender.capture(authority, {0, 0, 2, 0}).expect("capture transfer");
REQUIRE_EQ(parts.size(), std::size_t(3));
PixelWorld replica(8501);
CHECK_EQ(receiver.receive(parts[2], replica).expect("part 2").transfersApplied,
std::uint32_t(0));
CHECK_EQ(receiver.receive(parts[0], replica).expect("part 0").transfersApplied,
std::uint32_t(0));
CHECK(receiver.receive(parts[2], replica).expect("duplicate part 2").duplicate);
CHECK_EQ(replica.chunkCount(), std::size_t(0));
REQUIRE_EQ(sender.pendingParts().size(), std::size_t(3));
const auto completed = receiver.receive(sender.pendingParts()[1], replica)
.expect("retransmitted missing part");
CHECK_EQ(completed.transfersApplied, std::uint32_t(1));
CHECK_EQ(completed.ack.acknowledgedThrough, std::uint64_t(1));
CHECK_EQ(replica.getCell(2 * kPixelChunkSize + 1, 1),
authority.getCell(2 * kPixelChunkSize + 1, 1));
const auto released = sender.acknowledge(completed.ack).expect("application ACK");
CHECK_EQ(released.transfersReleased, std::uint32_t(1));
CHECK_EQ(sender.inFlightTransferCount(), std::size_t(0));
}
TEST_CASE("pixelworld_transport_buffers_transfer_order_and_retains_tombstones_until_ACK") {
PixelWorld authority(8502);
authority.setMaterial(1, 1, "stone");
authority.setMaterial(kPixelChunkSize + 1, 1, "water");
auto sender = ReliablePixelChunkSender::create(92).expect("create sender");
auto receiver = ReliablePixelChunkReceiver::create(92).expect("create receiver");
PixelWorld replica(8502);
const auto initial = sender.capture(authority, {0, 0, 1, 0}).expect("initial");
PixelChunkReceiveReceipt receipt;
for (const auto& part : initial)
receipt = receiver.receive(part, replica).expect("receive initial");
sender.acknowledge(receipt.ack).expect("ACK initial");
authority.setMaterial(2, 2, "oil");
const auto second = sender.capture(authority, {0, 0, 1, 0}).expect("second");
authority.setMaterial(kPixelChunkSize + 2, 2, "wood");
const auto third = sender.capture(authority, {0, 0, 1, 0}).expect("third");
REQUIRE_EQ(second.size(), std::size_t(1));
REQUIRE_EQ(third.size(), std::size_t(1));
CHECK_EQ(receiver.receive(third.front(), replica).expect("third early").transfersApplied,
std::uint32_t(0));
const auto ordered = receiver.receive(second.front(), replica).expect("fill transfer gap");
CHECK_EQ(ordered.transfersApplied, std::uint32_t(2));
CHECK_EQ(ordered.ack.acknowledgedThrough, std::uint64_t(3));
CHECK_EQ(replica.getCell(kPixelChunkSize + 2, 2),
authority.getCell(kPixelChunkSize + 2, 2));
sender.acknowledge(ordered.ack).expect("ACK ordered transfers");
const auto evicted = sender.capture(authority, {0, 0, 0, 0}).expect("evict interest");
REQUIRE_EQ(evicted.size(), std::size_t(1));
CHECK(evicted.front().batch.chunks.front().removed);
CHECK_EQ(sender.retainedTombstoneCount(), std::size_t(1));
const auto tombstone = receiver.receive(evicted.front(), replica).expect("apply tombstone");
CHECK_EQ(replica.getMaterial(kPixelChunkSize + 1, 1), int(MaterialId::Air));
sender.acknowledge(tombstone.ack).expect("ACK tombstone");
CHECK_EQ(sender.retainedTombstoneCount(), std::size_t(0));
CHECK_EQ(sender.acknowledge(tombstone.ack).expect("duplicate ACK").transfersReleased,
std::uint32_t(0));
}
TEST_CASE("pixelworld_transport_window_failure_does_not_advance_capture_cursor") {
PixelChunkTransportConfig config;
config.maximumInFlightTransfers = 1;
PixelWorld authority(8503);
authority.setMaterial(1, 1, "stone");
auto sender = ReliablePixelChunkSender::create(93, config).expect("create sender");
auto receiver = ReliablePixelChunkReceiver::create(93, config).expect("create receiver");
PixelWorld replica(8503);
const auto initial = sender.capture(authority, {0, 0, 0, 0}).expect("initial");
authority.setMaterial(2, 2, "water");
CHECK(!sender.capture(authority, {0, 0, 0, 0}).ok());
const auto first = receiver.receive(initial.front(), replica).expect("receive initial");
sender.acknowledge(first.ack).expect("free window");
const auto retry = sender.capture(authority, {0, 0, 0, 0}).expect("retry capture");
REQUIRE_EQ(retry.size(), std::size_t(1));
CHECK_EQ(retry.front().batch.sourceRevision, authority.revision());
CHECK_EQ(int(retry.front().batch.chunks.front().cells[2 * kPixelChunkSize + 2].material),
int(MaterialId::Water));
}
TEST_CASE("pixelworld_transport_rejects_conflicting_duplicate_and_invalid_ACK") {
PixelWorld authority(8504);
authority.setMaterial(1, 1, "stone");
authority.setMaterial(kPixelChunkSize + 1, 1, "water");
auto sender = ReliablePixelChunkSender::create(94).expect("create sender");
auto receiver = ReliablePixelChunkReceiver::create(94).expect("create receiver");
const auto parts = sender.capture(authority, {0, 0, 1, 0}).expect("capture");
REQUIRE_EQ(parts.size(), std::size_t(2));
PixelWorld replica(8504);
receiver.receive(parts.front(), replica).expect("receive partial");
auto conflict = parts.front();
conflict.batch.sourceRevision += 1;
CHECK(!receiver.receive(std::move(conflict), replica).ok());
CHECK(!sender.acknowledge({94, 1, authority.revision() + 1}).ok());
CHECK_EQ(sender.inFlightTransferCount(), std::size_t(1));
}
Reference in new issue View Git Blame Copy Permalink