feat(synapse-sdk): batch storage uploads by default

packages/synapse-core/src/sp/create-piece-batcher.ts

@@ -24,11 +24,18 @@ import { findPiece } from './find-piece.ts'
 import { waitForPullPieces } from './pull-pieces.ts'
 import { type UploadPieceStreamingData, uploadPieceStreaming } from './upload-streaming.ts'
 
-export type EnqueuePiece = LimiterPiece
+export type EnqueuePiece = LimiterPiece & {
+  /** Raw byte size when the piece was parked by `upload()`. */
+  size?: number
+}
 
 export type FlushResult = {
   txHash: Hex
+  confirmedTxHash?: Hex
   statusUrl: string
+  dataSetId: bigint
+  clientDataSetId: bigint
+  isNewDataSet: boolean
   pieces: EnqueuePiece[]
 }
 
@@ -37,6 +44,10 @@ export type PieceResult = FlushResult & {
   batchIndex: number
 }
 
+export type UploadResult = PieceResult & {
+  size: number
+}
+
 /**
  * When to flush a tumbling addPieces window (limiter overflow, `flush()`, and
  * `close()` always flush regardless).
@@ -56,7 +67,7 @@ export type PieceBatcherWait = { kind: 'delay'; ms: number } | { kind: 'limiter'
 export type OnParked = (piece: EnqueuePiece) => void | Promise<void>
 
 export type UploadInput = {
-  data: File | Uint8Array | ReadableStream<Uint8Array>
+  data: File | UploadPieceStreamingData
   /** Known length for a stream (`File` uses `.size`). */
   size?: number
   metadata?: MetadataObject
@@ -71,6 +82,8 @@ export type PullInput = {
   sourceUrl: string
   metadata?: MetadataObject
   onParked?: OnParked
+  onStatus?: (response: waitForPullPieces.ReturnType) => void
+  signal?: AbortSignal
 }
 
 type Slot = {
@@ -81,7 +94,7 @@ type Slot = {
 
 export type PieceBatcher = {
   /** Stream onto this SP, then join the addPieces window. */
-  upload: (input: UploadInput) => Promise<PieceResult>
+  upload: (input: UploadInput) => Promise<UploadResult>
   /** Pull onto this SP (own extraData), then join the addPieces window. */
   pull: (input: PullInput) => Promise<PieceResult>
   /** Already on this SP. Join the addPieces window only. */
@@ -236,19 +249,36 @@ export function createPieceBatcher(
     const pieces = batch.map((slot) => slot.piece)
 
     try {
-      const submitted =
-        dataSet == null
-          ? await flushCreate(pieces)
-          : await addPieces(client, {
-              serviceURL: serviceURL(),
-              dataSetId: dataSet.dataSetId,
-              clientDataSetId: dataSet.clientDataSetId,
-              pieces,
-            })
+      const isNewDataSet = dataSet == null
+      let submitted: {
+        txHash: Hex
+        confirmedTxHash?: Hex
+        statusUrl: string
+        dataSetId: bigint
+        clientDataSetId: bigint
+      }
+      if (dataSet == null) {
+        submitted = await flushCreate(pieces)
+      } else {
+        submitted = {
+          ...(await addPieces(client, {
+            serviceURL: serviceURL(),
+            dataSetId: dataSet.dataSetId,
+            clientDataSetId: dataSet.clientDataSetId,
+            pieces,
+          })),
+          dataSetId: dataSet.dataSetId,
+          clientDataSetId: dataSet.clientDataSetId,
+        }
+      }
 
       const result: FlushResult = {
         txHash: submitted.txHash,
+        ...(submitted.confirmedTxHash == null ? {} : { confirmedTxHash: submitted.confirmedTxHash }),
         statusUrl: submitted.statusUrl,
+        dataSetId: submitted.dataSetId,
+        clientDataSetId: submitted.clientDataSetId,
+        isNewDataSet,
         pieces,
       }
       for (const [batchIndex, slot] of batch.entries()) {
@@ -275,7 +305,13 @@ export function createPieceBatcher(
     }
   }
 
-  async function flushCreate(pieces: EnqueuePiece[]): Promise<{ txHash: Hex; statusUrl: string }> {
+  async function flushCreate(pieces: EnqueuePiece[]): Promise<{
+    txHash: Hex
+    confirmedTxHash?: Hex
+    statusUrl: string
+    dataSetId: bigint
+    clientDataSetId: bigint
+  }> {
     if (payee == null) {
       throw new ValidationError('`payee` is required when dataSet is undefined.')
     }
@@ -294,7 +330,16 @@ export function createPieceBatcher(
       throw new DataSetNotFoundError(created.dataSetId)
     }
     dataSet = resolved
-    return submitted
+    return {
+      txHash: submitted.txHash,
+      ...(created.confirmedTxHash == null ? {} : { confirmedTxHash: created.confirmedTxHash }),
+      statusUrl: new URL(
+        `/pdp/data-sets/${created.dataSetId}/pieces/added/${submitted.txHash}`,
+        serviceURL()
+      ).toString(),
+      dataSetId: created.dataSetId,
+      clientDataSetId: createClientDataSetId,
+    }
   }
 
   function cancelTimer(): void {
@@ -368,7 +413,7 @@ export function createPieceBatcher(
     return internalEnqueue(piece)
   }
 
-  async function upload(input: UploadInput): Promise<PieceResult> {
+  async function upload(input: UploadInput): Promise<UploadResult> {
     let data: UploadPieceStreamingData
     let size = input.size
     if (isUint8Array(input.data)) {
@@ -380,7 +425,8 @@ export function createPieceBatcher(
     } else {
       data = input.data
     }
-    return parkAndEnqueue(async () => {
+    let uploadedSize: number | undefined
+    const result = await parkAndEnqueue(async () => {
       if (input.pieceCid != null) {
         assertPieceCidSize(input.pieceCid)
       }
@@ -392,14 +438,19 @@ export function createPieceBatcher(
         onProgress: input.onProgress,
         signal: input.signal,
       })
+      uploadedSize = uploaded.size
       await findPiece({
         serviceURL: serviceURL(),
         pieceCid: uploaded.pieceCid,
         poll: true,
         signal: input.signal,
       })
-      return { pieceCid: uploaded.pieceCid, metadata: input.metadata }
+      return { pieceCid: uploaded.pieceCid, metadata: input.metadata, size: uploaded.size }
     }, input.onParked)
+    if (uploadedSize == null) {
+      throw new ValidationError('Piece upload completed without a size.')
+    }
+    return { ...result, size: uploadedSize }
   }
 
   async function pull(input: PullInput): Promise<PieceResult> {
@@ -430,13 +481,17 @@ export function createPieceBatcher(
               payer,
               cdn,
               metadata: datasetMetadata,
+              signal: input.signal,
+              onStatus: input.onStatus,
             })
           : await waitForPullPieces(client, {
               serviceURL: serviceURL(),
               pieces: [pullPiece],
               extraData,
               dataSetId: dataSet.dataSetId,
               clientDataSetId: dataSet.clientDataSetId,
+              signal: input.signal,
+              onStatus: input.onStatus,
             })
       if (pullResult.status === 'failed') {
         throw new PullError('Pull failed.')
@@ -453,15 +508,15 @@ export function createPieceBatcher(
   }
 
   async function flush(): Promise<FlushResult | undefined> {
+    await Promise.allSettled([...inFlight])
+    await Promise.resolve()
+    await scheduledIdle
     return lock(() => flushWindowInternal())
   }
 
   async function close(): Promise<void> {
     closed = true
-    await Promise.allSettled([...inFlight])
-    await Promise.resolve()
-    await scheduledIdle
-    await lock(() => flushWindowInternal())
+    await flush()
   }
 
   return {

packages/synapse-core/test/create-piece-batcher.test.ts

@@ -287,6 +287,28 @@ describe('createPieceBatcher', () => {
     assert.equal(b.txHash, mockTxHash2)
   })
 
+  it('should pass the active data set to a custom limiter', async () => {
+    server.use(addPiecesCaptureHandler(() => undefined))
+    let observedDataSet: PdpDataSet | undefined
+    const batcher = createPieceBatcher(client, {
+      dataSet: createDataSet(),
+      wait: { kind: 'limiter' },
+      limiter: (options) => {
+        if (options.kind !== 'addPieces') {
+          return false
+        }
+        observedDataSet = options.dataSet
+        return options.dataSet?.dataSetId === 1n
+      },
+    })
+
+    const pending = batcher.enqueue({ pieceCid: pieceCidA })
+    await batcher.close()
+    await pending
+
+    assert.equal(observedDataSet?.dataSetId, 1n)
+  })
+
   it('should mix upload and pull in one window', async () => {
     const addBodies: addPiecesApiRequest.RequestBody[] = []
     let pullExtraData: string | undefined
@@ -548,6 +570,7 @@ describe('createPieceBatcher', () => {
 
     assert.ok(batcher.dataSet)
     assert.equal(batcher.dataSet?.dataSetId, 1n)
+    assert.equal(batcher.dataSet?.provider.id, 1n)
     assert.equal(created.txHash, mockTxHash)
     assert.equal(added.txHash, mockTxHash2)
     assert.equal(addBodies.length, 1)

packages/synapse-sdk/src/storage/context.ts

@@ -48,6 +48,7 @@ import {
 } from '@filoz/synapse-core/utils'
 import {
   fetchProviderSelectionInput,
+  getPdpDataSet,
   metadataMatches,
   type ResolvedLocation,
   selectProviders,
@@ -80,6 +81,7 @@ import type {
 import { createError, SIZE_CONSTANTS } from '../utils/index.ts'
 import { combineMetadata } from '../utils/metadata.ts'
 import type { WarmStorageService } from '../warm-storage/index.ts'
+import { type BatchedUploadResult, getPieceBatchingService } from './piece-batching.ts'
 import { terminateServiceFlow } from './terminate.ts'
 
 const NO_REMAINING_PROVIDERS_ERROR_MESSAGE = 'No approved service providers available'
@@ -144,6 +146,25 @@ export class StorageContext {
     return this._dataSetId
   }
 
+  /** @internal Synchronize state after a shared piece batcher creates or resolves a data set. */
+  syncBatcherDataSet(dataSetId: bigint, clientDataSetId: bigint): void {
+    this._dataSetId = dataSetId
+    this._clientDataSetId = clientDataSetId
+  }
+
+  /** @internal Resolve the minimal data-set state required by a piece batcher. */
+  async getBatcherDataSet(): Promise<SP.PieceBatcher['dataSet']> {
+    if (this._dataSetId == null) {
+      return undefined
+    }
+    const dataSet = await getPdpDataSet(this._readClient, { dataSetId: this._dataSetId })
+    if (dataSet == null) {
+      throw createError('StorageContext', 'getBatcherDataSet', 'Data set not found')
+    }
+    this._clientDataSetId = dataSet.clientDataSetId
+    return dataSet
+  }
+
   private assertPiecesFitMessage(pieces: Array<{ pieceCid: PieceCID; metadata?: MetadataObject }>): void {
     SP.assertAddPiecesFit(
       this._dataSetId
@@ -1021,6 +1042,57 @@ export class StorageContext {
    * @returns Upload result with pieceCid, size, and a single-element copies array
    */
   async upload(data: UploadPieceStreamingData, options?: UploadOptions): Promise<UploadResult> {
+    const batching = getPieceBatchingService(this._synapse)
+    if (batching != null) {
+      let parked = false
+      const task = batching.upload(this, {
+        data,
+        pieceCid: options?.pieceCid,
+        metadata: options?.pieceMetadata,
+        signal: options?.signal,
+        onProgress: options?.onProgress,
+        onParked: (piece) => {
+          parked = true
+          options?.onStored?.(this._provider.id, piece.pieceCid)
+        },
+        onSubmitted: (submitted) =>
+          options?.onPiecesAdded?.(submitted.txHash, this._provider.id, [{ pieceCid: submitted.pieceCid }]),
+      })
+
+      let result: BatchedUploadResult
+      try {
+        result = await task.committed
+      } catch (error) {
+        throw createError(
+          'StorageContext',
+          parked ? 'commit' : 'store',
+          parked ? 'Failed to commit pieces on-chain' : 'Failed to store piece on service provider',
+          error
+        )
+      }
+
+      options?.onPiecesConfirmed?.(result.dataSetId, this._provider.id, [
+        { pieceId: result.pieceId, pieceCid: result.pieceCid },
+      ])
+      return {
+        pieceCid: result.pieceCid,
+        size: result.size,
+        requestedCopies: 1,
+        complete: true,
+        copies: [
+          {
+            providerId: this._provider.id,
+            dataSetId: result.dataSetId,
+            pieceId: result.pieceId,
+            role: 'primary',
+            retrievalUrl: this.getPieceUrl(result.pieceCid),
+            isNewDataSet: result.isNewDataSet,
+          },
+        ],
+        failedAttempts: [],
+      }
+    }
+
     // Store phase
     const storeResult = await this.store(data, {
       pieceCid: options?.pieceCid,