fix: datalake upload fixes (#9680)

Signed-off-by: Alexander Onnikov <Alexander.Onnikov@xored.com>

services/datalake/pod-datalake/src/datalake/datalake.ts

@@ -167,7 +167,7 @@ export class DatalakeImpl implements Datalake {
     const data = await this.db.getData(ctx, { hash, location })
 
     if (data !== null) {
-      // Lucky boy, nothing to upload, use existing blob
+      // Nothing to upload, use existing blob
       await this.db.createBlob(ctx, { workspace, name, hash, location })
 
       try {
@@ -189,25 +189,48 @@ export class DatalakeImpl implements Datalake {
         lastModified
       }
 
-      let data: Readable | Buffer = body
+      let dataToUpload: Readable | Buffer = body
+      let cacheBuffer: Buffer | undefined
 
       if (this.options.cache.enabled && size <= this.options.cache.blobSize) {
-        data = await streamToBuffer(body)
-        const entry: CacheEntry = {
-          body: data,
-          bodyLength: data.length,
-          bodyEtag: etag,
-          size,
+        cacheBuffer = await streamToBuffer(body)
+        dataToUpload = cacheBuffer
+      }
+
+      await bucket.put(ctx, filename, dataToUpload, putOptions)
+      const head = await bucket.head(ctx, filename)
+
+      if (head === null) {
+        ctx.error('failed to upload blob: uploaded blob not found', { workspace, name })
+        throw new Error('Failed to upload blob: uploaded blob not found')
+      }
+
+      if (head.size !== size) {
+        ctx.error('failed to upload blob: uploaded blob size mismatch', {
+          workspace,
           name,
+          expected: size,
+          uploaded: head.size
+        })
+        await bucket.delete(ctx, filename)
+        throw new Error(`Failed to upload blob: uploaded blob size mismatch, expected ${size}, got ${head.size}`)
+      }
+
+      await this.db.createBlobData(ctx, { workspace, name, hash, location, filename, size, type: contentType })
+
+      if (cacheBuffer !== undefined) {
+        const entry: CacheEntry = {
+          body: cacheBuffer,
+          bodyLength: cacheBuffer.length,
+          bodyEtag: head.etag,
+          size,
           etag,
+          name,
           ...putOptions
         }
         this.cache.set(hash, entry)
       }
 
-      await bucket.put(ctx, filename, data, putOptions)
-      await this.db.createBlobData(ctx, { workspace, name, hash, location, filename, size, type: contentType })
-
       try {
         const event =
           blob != null

services/datalake/pod-datalake/src/handlers/blob.ts

@@ -266,7 +266,13 @@ export async function handleUploadFormData (
           throw err
         }
 
-        const data = file.tempFilePath !== undefined ? fs.createReadStream(file.tempFilePath) : file.data
+        let data: Buffer | Readable = file.data
+        if (file.tempFilePath !== undefined) {
+          data = fs.createReadStream(file.tempFilePath)
+          data.on('error', (err) => {
+            ctx.error('stream error during upload', { workspace, name, file: file.name, error: err })
+          })
+        }
 
         try {
           const metadata = await datalake.put(ctx, workspace, name, sha256, data, {

services/datalake/pod-datalake/src/s3/bucket.ts

@@ -124,12 +124,7 @@ class S3BucketImpl implements S3Bucket {
     }
   }
 
-  async put (
-    ctx: MeasureContext,
-    key: string,
-    body: Readable | Buffer | string,
-    options: S3PutOptions
-  ): Promise<S3Object> {
+  async put (ctx: MeasureContext, key: string, body: Readable | Buffer | string, options: S3PutOptions): Promise<void> {
     const command = {
       Bucket: this.bucket,
       Key: key,
@@ -143,16 +138,7 @@ class S3BucketImpl implements S3Bucket {
     }
 
     if (Buffer.isBuffer(body)) {
-      const result = await ctx.with('s3.putObject', {}, () => this.client.putObject(command), { bucket: this.bucket })
-
-      return {
-        key,
-        etag: result.ETag ?? '',
-        size: result.Size ?? 0,
-        contentType: options.contentType,
-        lastModified: options.lastModified,
-        cacheControl: options.cacheControl
-      }
+      await ctx.with('s3.putObject', {}, () => this.client.putObject(command), { bucket: this.bucket })
     } else {
       const upload = new Upload({
         client: this.client,
@@ -161,16 +147,7 @@ class S3BucketImpl implements S3Bucket {
         leavePartsOnError: false
       })
 
-      const result = await ctx.with('s3.upload', {}, () => upload.done(), { bucket: this.bucket })
-
-      return {
-        key,
-        etag: result.ETag ?? '',
-        size: options.contentLength,
-        contentType: options.contentType,
-        lastModified: options.lastModified,
-        cacheControl: options.cacheControl
-      }
+      await ctx.with('s3.upload', {}, () => upload.done(), { bucket: this.bucket })
     }
   }
 

services/datalake/pod-datalake/src/s3/types.ts

@@ -66,7 +66,7 @@ export interface S3Bucket {
 
   head: (ctx: MeasureContext, key: string) => Promise<S3Object | null>
   get: (ctx: MeasureContext, key: string, options?: S3GetOptions) => Promise<S3ObjectBody | null>
-  put: (ctx: MeasureContext, key: string, body: Readable | Buffer | string, options: S3PutOptions) => Promise<S3Object>
+  put: (ctx: MeasureContext, key: string, body: Readable | Buffer | string, options: S3PutOptions) => Promise<void>
   delete: (ctx: MeasureContext, key: string) => Promise<void>
 
   // multipart