Fix owners in PDisk can write to other owner's chunks (#9869)

Author: va-kuznecov

ydb/core/blobstorage/pdisk/blobstorage_pdisk_impl.cpp

@@ -2892,7 +2892,16 @@ bool TPDisk::PreprocessRequest(TRequestBase *request) {
                  }
             }
             TChunkState &state = ChunkState[ev.ChunkIdx];
-            if (state.OwnerId == OwnerSystem) {
+            if (state.OwnerId != ev.Owner) {
+                err << "Can't write chunkIdx# " << ev.ChunkIdx
+                    << " chunk is owner by another owner."
+                    << " chunk's owner# " << state.OwnerId
+                    << " request's owner# " << ev.Owner;
+                SendChunkWriteError(ev, err.Str(), NKikimrProto::ERROR);
+                delete request;
+                return false;
+            }
+            if (!IsOwnerUser(state.OwnerId)) {
                 err << "Can't write chunkIdx# " << ev.ChunkIdx
                     << " destination chunk is owned by the system! ownerId# " << ev.Owner;
                 SendChunkWriteError(ev, err.Str(), NKikimrProto::ERROR);

ydb/core/blobstorage/pdisk/blobstorage_pdisk_ut.cpp

@@ -926,6 +926,50 @@ Y_UNIT_TEST_SUITE(TPDiskTest) {
         UNIT_ASSERT_VALUES_EQUAL(writeLog(), NKikimrProto::OK);
     }
 
+    Y_UNIT_TEST(TestChunkWriteCrossOwner) {
+        TActorTestContext testCtx({ false });
+
+        TVDiskMock vdisk1(&testCtx);
+        TVDiskMock vdisk2(&testCtx);
+
+        vdisk1.InitFull();
+        vdisk2.InitFull();
+
+        vdisk1.ReserveChunk();
+        vdisk2.ReserveChunk();
+
+        vdisk1.CommitReservedChunks();
+        vdisk2.CommitReservedChunks();
+
+        UNIT_ASSERT(vdisk1.Chunks[EChunkState::COMMITTED].size() == 1);
+        UNIT_ASSERT(vdisk2.Chunks[EChunkState::COMMITTED].size() == 1);
+
+        auto chunk1 = *vdisk1.Chunks[EChunkState::COMMITTED].begin();
+        auto chunk2 = *vdisk2.Chunks[EChunkState::COMMITTED].begin();
+
+        TString data(123, '0');
+        auto parts = MakeIntrusive<NPDisk::TEvChunkWrite::TStrokaBackedUpParts>(data);
+
+        // write to own chunk is OK
+        testCtx.TestResponse<NPDisk::TEvChunkWriteResult>(new NPDisk::TEvChunkWrite(
+            vdisk1.PDiskParams->Owner, vdisk1.PDiskParams->OwnerRound,
+            chunk1, 0, parts, nullptr, false, 0),
+            NKikimrProto::OK);
+        testCtx.TestResponse<NPDisk::TEvChunkWriteResult>(new NPDisk::TEvChunkWrite(
+            vdisk2.PDiskParams->Owner, vdisk2.PDiskParams->OwnerRound,
+            chunk2, 0, parts, nullptr, false, 0),
+            NKikimrProto::OK);
+
+        // write to neighbour's chunk is ERROR
+        testCtx.TestResponse<NPDisk::TEvChunkWriteResult>(new NPDisk::TEvChunkWrite(
+            vdisk1.PDiskParams->Owner, vdisk1.PDiskParams->OwnerRound,
+            chunk2, 0, parts, nullptr, false, 0),
+            NKikimrProto::ERROR);
+        testCtx.TestResponse<NPDisk::TEvChunkWriteResult>(new NPDisk::TEvChunkWrite(
+            vdisk2.PDiskParams->Owner, vdisk2.PDiskParams->OwnerRound,
+            chunk1, 0, parts, nullptr, false, 0),
+            NKikimrProto::ERROR);
+    }
 }
 
 Y_UNIT_TEST_SUITE(PDiskCompatibilityInfo) {
@@ -961,7 +1005,7 @@ Y_UNIT_TEST_SUITE(PDiskCompatibilityInfo) {
 
     void TestMajorVerionMigration(TCurrent oldInfo, TCurrent intermediateInfo, TCurrent newInfo) {
         TCompatibilityInfoTest::Reset(&oldInfo);
-    
+
         TActorTestContext testCtx({
             .IsBad = false,
             .SuppressCompatibilityCheck = false,