From ec9387260890569940a15b2cff65d3f0403adc8a Mon Sep 17 00:00:00 2001 From: Alexander Onnikov Date: Fri, 15 Aug 2025 15:52:39 +0700 Subject: [PATCH] fix: datalake upload fixes (#9680) Signed-off-by: Alexander Onnikov --- .../pod-datalake/src/datalake/datalake.ts | 45 ++++++++++++++----- .../pod-datalake/src/handlers/blob.ts | 8 +++- .../datalake/pod-datalake/src/s3/bucket.ts | 29 ++---------- .../datalake/pod-datalake/src/s3/types.ts | 2 +- 4 files changed, 45 insertions(+), 39 deletions(-) diff --git a/services/datalake/pod-datalake/src/datalake/datalake.ts b/services/datalake/pod-datalake/src/datalake/datalake.ts index 68c90edc90..8fe70d8254 100644 --- a/services/datalake/pod-datalake/src/datalake/datalake.ts +++ b/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 diff --git a/services/datalake/pod-datalake/src/handlers/blob.ts b/services/datalake/pod-datalake/src/handlers/blob.ts index 0061591cb6..f90d49dccc 100644 --- a/services/datalake/pod-datalake/src/handlers/blob.ts +++ b/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, { diff --git a/services/datalake/pod-datalake/src/s3/bucket.ts b/services/datalake/pod-datalake/src/s3/bucket.ts index 00af00130b..1d877793d5 100644 --- a/services/datalake/pod-datalake/src/s3/bucket.ts +++ b/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 { + async put (ctx: MeasureContext, key: string, body: Readable | Buffer | string, options: S3PutOptions): Promise { 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 }) } } diff --git a/services/datalake/pod-datalake/src/s3/types.ts b/services/datalake/pod-datalake/src/s3/types.ts index b1f28dc3fb..eb3968da9a 100644 --- a/services/datalake/pod-datalake/src/s3/types.ts +++ b/services/datalake/pod-datalake/src/s3/types.ts @@ -66,7 +66,7 @@ export interface S3Bucket { head: (ctx: MeasureContext, key: string) => Promise get: (ctx: MeasureContext, key: string, options?: S3GetOptions) => Promise - put: (ctx: MeasureContext, key: string, body: Readable | Buffer | string, options: S3PutOptions) => Promise + put: (ctx: MeasureContext, key: string, body: Readable | Buffer | string, options: S3PutOptions) => Promise delete: (ctx: MeasureContext, key: string) => Promise // multipart