#include "zeroerr/assert.h" #include "zeroerr/unittest.h" #include "pixelworld/PixelWorld.h" #include "pixelworld/streaming/PixelWorldStreaming.h" #include "pixelworld/streaming/PixelWorldTransport.h" #include #include #include using namespace eve::pixelworld; using namespace eve::pixelworld_streaming; namespace { eve::SnapshotHashProvider archiveHashProvider() { return [](std::string_view input) -> eve::Result { 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::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()->erase("sourceLastEditSequence"); legacyPayload.getIf()->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()->at("chunks").getIf(); auto& stored = *chunks.front().getIf()->at("stored").getIf(); 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)); }