[IN-115] feat(size-check): check size when upload finished

lib/core/buckets/usecase.ts

@@ -43,6 +43,20 @@ export class BucketNotFoundError extends Error {
     Object.setPrototypeOf(this, BucketNotFoundError.prototype);
   }
 }
+export class UploadSizeDoesNotMatchError extends Error {
+  constructor(message?: string) {
+    super(message ?? `Size does not match`);
+
+    Object.setPrototypeOf(this, UploadSizeDoesNotMatchError.prototype);
+  }
+}
+export class UploadNotFoundInStorageError extends Error {
+  constructor() {
+    super(`Upload not found in storage`);
+
+    Object.setPrototypeOf(this, UploadNotFoundInStorageError.prototype);
+  }
+}
 
 export class BucketEntryFrameNotFoundError extends Error {
   constructor() {
@@ -664,11 +678,8 @@ export class BucketsUsecase {
       if (contact.objectCheckNotRequired) {
         contactsThatStoreTheShard.push(contact);
       } else {
-        const storesObject = await StorageGateway.stores(contact, uuid);
-
-        if (storesObject) {
-          contactsThatStoreTheShard.push(contact);
-        }
+        await this.validateObjectInStorage(contact, uuid, data_size);
+        contactsThatStoreTheShard.push(contact);
       }
     }
 
@@ -748,4 +759,25 @@ export class BucketsUsecase {
 
     return this.bucketsRepository.create(payload);
   }
+
+    async validateObjectInStorage(
+    contact: Contact,
+    uuid: string,
+    expectedSize: number
+    ): Promise<void> {
+
+        const objectMeta = await StorageGateway.getMeta(contact, uuid);
+
+        if (!objectMeta) {
+            throw new UploadNotFoundInStorageError();
+        }
+
+        if (objectMeta.size !== expectedSize) {
+            console.warn(`validateObjectInStorage | Size mismatch for contact ${contact.id} object ${uuid}. Expected: ${expectedSize}, Got: ${objectMeta.size}`);
+
+            throw new UploadSizeDoesNotMatchError(
+            `Size does not match. Expected: ${expectedSize}, Got: ${objectMeta.size}`
+            );
+        }
+    }
 }

lib/core/storage/StorageGateway.ts

@@ -19,6 +19,23 @@ export class StorageGateway {
     });
   }
 
+
+  static async getMeta(contact: Contact, objectKey: string): Promise<{size: number} | null> {
+    const { address, port } = contact;
+
+    const httpUrl = `http://${address}:${port}/v2/shard/${objectKey}/meta`;
+
+    try {
+      const response = await axios.get(httpUrl);
+      return response.data;
+    } catch (err) {
+      if (axios.isAxiosError(err) && err.response?.status === 404) {
+        return null;
+      }
+      throw err;
+    }
+  }
+
   static async getLinks(contact: Contact, objectKeys: string[]): Promise<string[]> {
     const { address, port } = contact;
 

lib/server/routes/buckets.js

@@ -22,7 +22,7 @@ const { v4: uuidv4, validate: uuidValidate } = require('uuid');
 const { isHexString } = require('../middleware/farmer-auth');
 const axios = require('axios');
 const { MongoDBBucketEntriesRepository } = require('../../core/bucketEntries/MongoDBBucketEntriesRepository');
-const { BucketsUsecase, BucketEntryNotFoundError, BucketEntryFrameNotFoundError, BucketNotFoundError, BucketForbiddenError, MissingUploadsError, MaxSpaceUsedError, InvalidUploadIndexes, InvalidMultiPartValueError, NoNodeFoundError, EmptyMirrorsError, BucketNameAlreadyInUse } = require('../../core/buckets/usecase');
+const { BucketsUsecase, BucketEntryNotFoundError, BucketEntryFrameNotFoundError, BucketNotFoundError, BucketForbiddenError, MissingUploadsError, MaxSpaceUsedError, InvalidUploadIndexes, InvalidMultiPartValueError, NoNodeFoundError, EmptyMirrorsError, BucketNameAlreadyInUse, UploadSizeDoesNotMatchError, UploadNotFoundInStorageError } = require('../../core/buckets/usecase');
 const { BucketEntriesUsecase, BucketEntryVersionNotFoundError } = require('../../core/bucketEntries/usecase');
 const { MongoDBBucketEntryShardsRepository } = require('../../core/bucketEntryShards/MongoDBBucketEntryShardsRepository');
 const { MongoDBMirrorsRepository } = require('../../core/mirrors/MongoDBMirrorsRepository');
@@ -1334,7 +1334,19 @@ BucketsRouter.prototype.finishUpload = async function (req, res, next) {
       requestHeaders: err?.config?.headers
     };
 
-    log.error('finishUpload: Error for bucket %s: for user: %s, details: %s', bucketId, req.user.uuid, JSON.stringify(errorDetails));
+    if (err instanceof UploadSizeDoesNotMatchError) {
+      log.error('[finishUpload][SizeDoesNotMatchError] Error for bucket %s: for user: %s, details: %s', bucketId, req.user.uuid, JSON.stringify(errorDetails));
+
+      return next(new errors.ConflictError(err.message));
+    }
+
+    if (err instanceof UploadNotFoundInStorageError) {
+      log.error('[finishUpload][UploadNotFoundInStorageError] Error for bucket %s: for user: %s, details: %s', bucketId, req.user.uuid, JSON.stringify(errorDetails));
+
+      return next(new errors.NotFoundError(err.message));
+    }
+
+    log.error('[finishUpload]: Error for bucket %s: for user: %s, details: %s', bucketId, req.user.uuid, JSON.stringify(errorDetails));
 
     return next(new errors.InternalError(err.message));
   }

tests/lib/core/buckets/usecase.test.ts

@@ -10,7 +10,13 @@ import { UsersRepository } from '../../../../lib/core/users/Repository';
 
 import { MongoDBBucketsRepository } from '../../../../lib/core/buckets/MongoDBBucketsRepository';
 import { MongoDBBucketEntriesRepository } from '../../../../lib/core/bucketEntries/MongoDBBucketEntriesRepository';
-import { BucketNameAlreadyInUse, BucketNotFoundError, BucketsUsecase } from '../../../../lib/core/buckets/usecase';
+import {
+  BucketNameAlreadyInUse,
+  BucketNotFoundError,
+  BucketsUsecase,
+  UploadNotFoundInStorageError,
+  UploadSizeDoesNotMatchError,
+} from '../../../../lib/core/buckets/usecase';
 import { MongoDBFramesRepository } from '../../../../lib/core/frames/MongoDBFramesRepository';
 import { MongoDBMirrorsRepository } from '../../../../lib/core/mirrors/MongoDBMirrorsRepository';
 import { MongoDBShardsRepository } from '../../../../lib/core/shards/MongoDBShardsRepository';
@@ -311,6 +317,58 @@ describe('findAllByUserAndCreatedSince()', () => {
     expect(result).toStrictEqual(buckets);
   });
 });
+  describe('validateObjectInStorage()', () => {
+    it('When called, then it should fetch the object metadata using the given contact and uuid', async () => {
+      const contact = fixtures.getContact();
+      const uuid = 'object-uuid-123';
+      const expectedSize = 512;
+
+      const getMetaStub = stub(StorageGateway, 'getMeta').resolves({ size: expectedSize });
+
+      await bucketsUsecase.validateObjectInStorage(contact, uuid, expectedSize);
+
+      expect(getMetaStub.calledOnce).toBeTruthy();
+      expect(getMetaStub.firstCall.args).toStrictEqual([contact, uuid]);
+    });
+
+    it('When object exists and size matches, then it should resolve successfully', async () => {
+      const contact = fixtures.getContact();
+      const uuid = 'object-uuid-123';
+      const expectedSize = 1024;
+
+      stub(StorageGateway, 'getMeta').resolves({ size: expectedSize });
+
+      await expect(
+        bucketsUsecase.validateObjectInStorage(contact, uuid, expectedSize)
+      ).resolves.toBeUndefined();
+    });
+
+    it('When getMeta returns null, then it should throw', async () => {
+      const contact = fixtures.getContact();
+      const uuid = 'missing-object';
+      const expectedSize = 512;
+
+      stub(StorageGateway, 'getMeta').resolves(null);
+
+      await expect(
+        bucketsUsecase.validateObjectInStorage(contact, uuid, expectedSize)
+      ).rejects.toThrow(UploadNotFoundInStorageError);
+    });
+
+    it('When size does not match, then it should throw', async () => {
+      const contact = fixtures.getContact();
+      const uuid = 'object-uuid-456';
+      const expectedSize = 1024;
+      const actualSize = 2048;
+
+      stub(StorageGateway, 'getMeta').resolves({ size: actualSize });
+
+      await expect(
+        bucketsUsecase.validateObjectInStorage(contact, uuid, expectedSize)
+      ).rejects.toThrow(UploadSizeDoesNotMatchError);
+    });
+  });
+
   describe('create()', () => {
     const user = fixtures.getUser();
 

tests/lib/core/storage/StorageGateway.test.ts

@@ -0,0 +1,85 @@
+import axios from 'axios';
+import { StorageGateway } from '../../../../lib/core/storage/StorageGateway';
+import fixtures from '../fixtures';
+import { v4 } from 'uuid';
+
+jest.mock('axios');
+
+const mockedAxios = axios as jest.Mocked<typeof axios>;
+
+describe('StorageGateway', () => {
+  beforeEach(() => {
+    jest.resetAllMocks();
+  });
+
+  describe('getMeta()', () => {
+    it('When the request succeeds, then it should return response.data', async () => {
+      const contact = fixtures.getContact({ address: 'storage.example.com', port: 3000 });
+      const objectKey = v4();
+      const meta = { size: 1024 };
+
+      mockedAxios.get.mockResolvedValueOnce({ data: meta });
+
+      const result = await StorageGateway.getMeta(contact, objectKey);
+
+      expect(result).toStrictEqual(meta);
+    });
+
+    it('When called, then it should construct the correct URL from contact address, port and objectKey', async () => {
+      const contact = fixtures.getContact({ address: 'mynode.example.com', port: 4567 });
+      const objectKey = v4();
+
+      mockedAxios.get.mockResolvedValueOnce({ data: { size: 512 } });
+
+      await StorageGateway.getMeta(contact, objectKey);
+
+      expect(mockedAxios.get).toHaveBeenCalledWith(
+        `http://mynode.example.com:4567/v2/shard/${objectKey}/meta`
+      );
+    });
+
+    it('When the storage node returns a 404, then it should return null', async () => {
+      const contact = fixtures.getContact();
+      const objectKey = v4();
+
+      const axiosError = Object.assign(new Error('Not Found'), {
+        isAxiosError: true,
+        response: { status: 404 },
+      });
+
+      mockedAxios.get.mockRejectedValueOnce(axiosError);
+      mockedAxios.isAxiosError.mockReturnValueOnce(true);
+
+      const result = await StorageGateway.getMeta(contact, objectKey);
+
+      expect(result).toBeNull();
+    });
+
+    it('When the storage node returns a non-404 error, then it should re-throw the error', async () => {
+      const contact = fixtures.getContact();
+      const objectKey = v4();
+
+      const axiosError = Object.assign(new Error('Internal Server Error'), {
+        isAxiosError: true,
+        response: { status: 500 },
+      });
+
+      mockedAxios.get.mockRejectedValueOnce(axiosError);
+      mockedAxios.isAxiosError.mockReturnValueOnce(true);
+
+      await expect(StorageGateway.getMeta(contact, objectKey)).rejects.toThrow('Internal Server Error');
+    });
+
+    it('When a non-axios error occurs, then it should re-throw the error', async () => {
+      const contact = fixtures.getContact();
+      const objectKey = v4();
+
+      const networkError = new Error('Network failure');
+
+      mockedAxios.get.mockRejectedValueOnce(networkError);
+      mockedAxios.isAxiosError.mockReturnValueOnce(false);
+
+      await expect(StorageGateway.getMeta(contact, objectKey)).rejects.toThrow('Network failure');
+    });
+  });
+});