feat(pdp): new PieceCID-last flow (internal only)

First pass at supporting the new flow @ filecoin-project/curio#668

Second pass at this would expose the stream up through the SDK so you're not
forced to pass down byte arrays. For now, this is just internal and hides the
details outside of PDPServer.

Author: rvagg

packages/synapse-sdk/src/pdp/server.ts

@@ -27,7 +27,7 @@
  */
 
 import { ethers } from 'ethers'
-import { asPieceCID, calculate as calculatePieceCID, downloadAndValidate } from '../piece/index.ts'
+import { asPieceCID, createPieceCIDStream, downloadAndValidate } from '../piece/index.ts'
 import type { DataSetData, MetadataEntry, PieceCID } from '../types.ts'
 import { validateDataSetMetadata, validatePieceMetadata } from '../utils/metadata.ts'
 import { constructFindPieceUrl, constructPieceUrl } from '../utils/piece.ts'
@@ -442,38 +442,23 @@ export class PDPServer {
   async uploadPiece(data: Uint8Array | ArrayBuffer): Promise<UploadResponse> {
     // Convert ArrayBuffer to Uint8Array if needed
     const uint8Data = data instanceof ArrayBuffer ? new Uint8Array(data) : data
-
-    // Calculate PieceCID
-    performance.mark('synapse:calculatePieceCID-start')
-    const pieceCid = calculatePieceCID(uint8Data)
-    performance.mark('synapse:calculatePieceCID-end')
-    performance.measure('synapse:calculatePieceCID', 'synapse:calculatePieceCID-start', 'synapse:calculatePieceCID-end')
     const size = uint8Data.length
 
-    const requestBody = {
-      pieceCid: pieceCid.toString(),
-      // No notify URL needed
-    }
-
-    // Create upload session or check if piece exists
-    performance.mark('synapse:POST.pdp.piece-start')
-    const createResponse = await fetch(`${this._serviceURL}/pdp/piece`, {
+    // Create upload session
+    performance.mark('synapse:POST.pdp.piece.uploads-start')
+    const createResponse = await fetch(`${this._serviceURL}/pdp/piece/uploads`, {
       method: 'POST',
       headers: {
         'Content-Type': 'application/json',
       },
-      body: JSON.stringify(requestBody),
+      body: JSON.stringify({}),
     })
-    performance.mark('synapse:POST.pdp.piece-end')
-    performance.measure('synapse:POST.pdp.piece', 'synapse:POST.pdp.piece-start', 'synapse:POST.pdp.piece-end')
-
-    if (createResponse.status === 200) {
-      // Piece already exists on server
-      return {
-        pieceCid,
-        size,
-      }
-    }
+    performance.mark('synapse:POST.pdp.piece.uploads-end')
+    performance.measure(
+      'synapse:POST.pdp.piece.uploads',
+      'synapse:POST.pdp.piece.uploads-start',
+      'synapse:POST.pdp.piece.uploads-end'
+    )
 
     if (createResponse.status !== 201) {
       const errorText = await createResponse.text()
@@ -489,37 +474,80 @@ export class PDPServer {
     }
 
     // Validate the location format and extract UUID
-    // Match /pdp/piece/upload/UUID or /piece/upload/UUID anywhere in the path
-    const locationMatch = location.match(/\/(?:pdp\/)?piece\/upload\/([a-fA-F0-9-]+)/)
+    // Match /pdp/piece/uploads/UUID anywhere in the path
+    const locationMatch = location.match(/\/pdp\/piece\/uploads\/([a-fA-F0-9-]+)/)
     if (locationMatch == null) {
       throw new Error(`Invalid Location header format: ${location}`)
     }
 
     const uploadUuid = locationMatch[1] // Extract just the UUID
 
-    // Upload the data
-    performance.mark('synapse:PUT.pdp.piece.upload-start')
-    const uploadResponse = await fetch(`${this._serviceURL}/pdp/piece/upload/${uploadUuid}`, {
+    // Create streaming CommP calculator
+    const { stream: pieceCidStream, getPieceCID } = createPieceCIDStream()
+
+    // Convert Uint8Array to ReadableStream for streaming through the CommP calculator
+    const dataStream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(uint8Data)
+        controller.close()
+      },
+    })
+
+    // Upload the data while calculating CommP in parallel
+    performance.mark('synapse:PUT.pdp.piece.uploads-start')
+    const uploadResponse = await fetch(`${this._serviceURL}/pdp/piece/uploads/${uploadUuid}`, {
       method: 'PUT',
       headers: {
         'Content-Type': 'application/octet-stream',
         'Content-Length': uint8Data.length.toString(),
-        // No Authorization header needed
       },
-      body: uint8Data,
-    })
-    performance.mark('synapse:PUT.pdp.piece.upload-end')
+      body: dataStream.pipeThrough(pieceCidStream),
+      duplex: 'half',
+    } as RequestInit)
+    performance.mark('synapse:PUT.pdp.piece.uploads-end')
     performance.measure(
-      'synapse:PUT.pdp.piece.upload',
-      'synapse:PUT.pdp.piece.upload-start',
-      'synapse:PUT.pdp.piece.upload-end'
+      'synapse:PUT.pdp.piece.uploads',
+      'synapse:PUT.pdp.piece.uploads-start',
+      'synapse:PUT.pdp.piece.uploads-end'
     )
 
     if (uploadResponse.status !== 204) {
       const errorText = await uploadResponse.text()
       throw new Error(`Failed to upload piece: ${uploadResponse.status} ${uploadResponse.statusText} - ${errorText}`)
     }
 
+    // Get the calculated PieceCID (available after stream completes)
+    const pieceCid = getPieceCID()
+    if (pieceCid == null) {
+      throw new Error('Failed to calculate PieceCID during upload')
+    }
+
+    // Finalize the upload with CommP validation
+    performance.mark('synapse:POST.pdp.piece.uploads.finalize-start')
+    const finalizeResponse = await fetch(`${this._serviceURL}/pdp/piece/uploads/${uploadUuid}`, {
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json',
+      },
+      body: JSON.stringify({
+        pieceCid: pieceCid.toString(),
+        size,
+      }),
+    })
+    performance.mark('synapse:POST.pdp.piece.uploads.finalize-end')
+    performance.measure(
+      'synapse:POST.pdp.piece.uploads.finalize',
+      'synapse:POST.pdp.piece.uploads.finalize-start',
+      'synapse:POST.pdp.piece.uploads.finalize-end'
+    )
+
+    if (finalizeResponse.status !== 200) {
+      const errorText = await finalizeResponse.text()
+      throw new Error(
+        `Failed to finalize upload: ${finalizeResponse.status} ${finalizeResponse.statusText} - ${errorText}`
+      )
+    }
+
     return {
       pieceCid,
       size,

packages/synapse-sdk/src/test/pdp-server.test.ts

@@ -502,26 +502,23 @@ describe('PDPServer', () => {
       const mockUuid = '12345678-90ab-cdef-1234-567890abcdef'
 
       server.use(
-        http.post<Record<string, never>, { pieceCid: string }>('http://pdp.local/pdp/piece', async ({ request }) => {
-          try {
-            const body = await request.json()
-            assert.exists(body.pieceCid)
-            return HttpResponse.text('Created', {
-              status: 201,
-              headers: {
-                Location: `/pdp/piece/upload/${mockUuid}`,
-              },
-            })
-          } catch (error) {
-            return HttpResponse.text((error as Error).message, {
-              status: 400,
-            })
-          }
+        http.post('http://pdp.local/pdp/piece/uploads', async () => {
+          return HttpResponse.text('Created', {
+            status: 201,
+            headers: {
+              Location: `/pdp/piece/uploads/${mockUuid}`,
+            },
+          })
         }),
-        http.put('http://pdp.local/pdp/piece/upload/:uuid', async () => {
+        http.put('http://pdp.local/pdp/piece/uploads/:uuid', async () => {
           return HttpResponse.text('No Content', {
             status: 204,
           })
+        }),
+        http.post('http://pdp.local/pdp/piece/uploads/:uuid', async () => {
+          return HttpResponse.text('OK', {
+            status: 200,
+          })
         })
       )
 
@@ -537,26 +534,23 @@ describe('PDPServer', () => {
       const mockUuid = 'fedcba09-8765-4321-fedc-ba0987654321'
 
       server.use(
-        http.post<Record<string, never>, { pieceCid: string }>('http://pdp.local/pdp/piece', async ({ request }) => {
-          try {
-            const body = await request.json()
-            assert.exists(body.pieceCid)
-            return HttpResponse.text('Created', {
-              status: 201,
-              headers: {
-                Location: `/pdp/piece/upload/${mockUuid}`,
-              },
-            })
-          } catch (error) {
-            return HttpResponse.text((error as Error).message, {
-              status: 400,
-            })
-          }
+        http.post('http://pdp.local/pdp/piece/uploads', async () => {
+          return HttpResponse.text('Created', {
+            status: 201,
+            headers: {
+              Location: `/pdp/piece/uploads/${mockUuid}`,
+            },
+          })
         }),
-        http.put('http://pdp.local/pdp/piece/upload/:uuid', async () => {
+        http.put('http://pdp.local/pdp/piece/uploads/:uuid', async () => {
           return HttpResponse.text('No Content', {
             status: 204,
           })
+        }),
+        http.post('http://pdp.local/pdp/piece/uploads/:uuid', async () => {
+          return HttpResponse.text('OK', {
+            status: 200,
+          })
         })
       )
 
@@ -565,33 +559,42 @@ describe('PDPServer', () => {
       assert.equal(result.size, 5)
     })
 
-    it('should handle existing piece (200 response)', async () => {
+    it('should throw on create upload session error', async () => {
       const testData = new Uint8Array([1, 2, 3, 4, 5])
-      const mockPieceCid = 'bafkzcibcd4bdomn3tgwgrh3g532zopskstnbrd2n3sxfqbze7rxt7vqn7veigmy'
 
       server.use(
-        http.post<Record<string, never>, { pieceCid: string }>('http://pdp.local/pdp/piece', async () => {
-          return HttpResponse.json(
-            { pieceCid: mockPieceCid },
-            {
-              status: 200,
-            }
-          )
+        http.post('http://pdp.local/pdp/piece/uploads', async () => {
+          return HttpResponse.text('Database error', {
+            status: 500,
+          })
         })
       )
 
-      // Should not throw - existing piece is OK
-      const result = await pdpServer.uploadPiece(testData)
-      assert.exists(result.pieceCid)
-      assert.equal(result.size, 5)
+      try {
+        await pdpServer.uploadPiece(testData)
+        assert.fail('Should have thrown error')
+      } catch (error: any) {
+        assert.include(error.message, 'Failed to create upload session')
+        assert.include(error.message, '500')
+        assert.include(error.message, 'Database error')
+      }
     })
 
-    it('should throw on create upload session error', async () => {
+    it('should throw on upload error', async () => {
       const testData = new Uint8Array([1, 2, 3, 4, 5])
+      const mockUuid = '12345678-90ab-cdef-1234-567890abcdef'
 
       server.use(
-        http.post<Record<string, never>, { pieceCid: string }>('http://pdp.local/pdp/piece', async () => {
-          return HttpResponse.text('Database error', {
+        http.post('http://pdp.local/pdp/piece/uploads', async () => {
+          return HttpResponse.text('Created', {
+            status: 201,
+            headers: {
+              Location: `/pdp/piece/uploads/${mockUuid}`,
+            },
+          })
+        }),
+        http.put('http://pdp.local/pdp/piece/uploads/:uuid', async () => {
+          return HttpResponse.text('Upload failed', {
             status: 500,
           })
         })
@@ -601,9 +604,44 @@ describe('PDPServer', () => {
         await pdpServer.uploadPiece(testData)
         assert.fail('Should have thrown error')
       } catch (error: any) {
-        assert.include(error.message, 'Failed to create upload session')
+        assert.include(error.message, 'Failed to upload piece')
         assert.include(error.message, '500')
-        assert.include(error.message, 'Database error')
+        assert.include(error.message, 'Upload failed')
+      }
+    })
+
+    it('should throw on finalization error', async () => {
+      const testData = new Uint8Array([1, 2, 3, 4, 5])
+      const mockUuid = '12345678-90ab-cdef-1234-567890abcdef'
+
+      server.use(
+        http.post('http://pdp.local/pdp/piece/uploads', async () => {
+          return HttpResponse.text('Created', {
+            status: 201,
+            headers: {
+              Location: `/pdp/piece/uploads/${mockUuid}`,
+            },
+          })
+        }),
+        http.put('http://pdp.local/pdp/piece/uploads/:uuid', async () => {
+          return HttpResponse.text('No Content', {
+            status: 204,
+          })
+        }),
+        http.post('http://pdp.local/pdp/piece/uploads/:uuid', async () => {
+          return HttpResponse.text('CommP mismatch', {
+            status: 400,
+          })
+        })
+      )
+
+      try {
+        await pdpServer.uploadPiece(testData)
+        assert.fail('Should have thrown error')
+      } catch (error: any) {
+        assert.include(error.message, 'Failed to finalize upload')
+        assert.include(error.message, '400')
+        assert.include(error.message, 'CommP mismatch')
       }
     })
   })