feat: allow custom shard size (#185)

Bubbles up the shard size option so `uploadFile` and `uploadDirectory` can set it.

packages/upload-client/README.md

@@ -150,7 +150,8 @@ function uploadDirectory(
  options: {
  retries?: number
  signal?: AbortSignal
- onShardStored: ShardStoredCallback
+ onShardStored?: ShardStoredCallback
+ shardSize?: number
  } = {}
 ): Promise<CID>
 ```
@@ -170,7 +171,8 @@ function uploadFile(
  options: {
  retries?: number
  signal?: AbortSignal
- onShardStored: ShardStoredCallback
+ onShardStored?: ShardStoredCallback
+ shardSize?: number
  } = {}
 ): Promise<CID>
 ```

packages/upload-client/src/index.js

@@ -7,11 +7,6 @@ import { ShardingStream, ShardStoringStream } from './sharding.js'
 export { Storage, Upload, UnixFS, CAR }
 export * from './sharding.js'
 
-/**
- * @typedef {(meta: import('./types').CARMetadata) => void} StoredShardCallback
- * @typedef {import('./types').RequestOptions & { onStoredShard?: StoredShardCallback }} UploadOptions
- */
-
 /**
  * Uploads a file to the service and returns the root data CID for the
  * generated DAG.
@@ -32,7 +27,7 @@ export * from './sharding.js'
  *
  * The issuer needs the `store/add` and `upload/add` delegated capability.
  * @param {Blob} file File data.
- * @param {UploadOptions} [options]
+ * @param {import('./types').UploadOptions} [options]
  */
 export async function uploadFile(conf, file, options = {}) {
  return await uploadBlockStream(
@@ -63,7 +58,7 @@ export async function uploadFile(conf, file, options = {}) {
  *
  * The issuer needs the `store/add` and `upload/add` delegated capability.
  * @param {import('./types').FileLike[]} files File data.
- * @param {UploadOptions} [options]
+ * @param {import('./types').UploadOptions} [options]
  */
 export async function uploadDirectory(conf, files, options = {}) {
  return await uploadBlockStream(
@@ -76,7 +71,7 @@ export async function uploadDirectory(conf, files, options = {}) {
 /**
  * @param {import('./types').InvocationConfig} conf
  * @param {ReadableStream<import('@ipld/unixfs').Block>} blocks
- * @param {UploadOptions} [options]
+ * @param {import('./types').UploadOptions} [options]
  * @returns {Promise<import('./types').AnyLink>}
  */
 async function uploadBlockStream(conf, blocks, options = {}) {
@@ -85,7 +80,7 @@ async function uploadBlockStream(conf, blocks, options = {}) {
  /** @type {import('./types').AnyLink?} */
  let root = null
  await blocks
- .pipeThrough(new ShardingStream())
+ .pipeThrough(new ShardingStream(options))
  .pipeThrough(new ShardStoringStream(conf, options))
  .pipeTo(
  new WritableStream({

packages/upload-client/src/sharding.js

@@ -13,9 +13,7 @@ const CONCURRENT_UPLOADS = 3
  */
 export class ShardingStream extends TransformStream {
  /**
- * @param {object} [options]
- * @param {number} [options.shardSize] The target shard size. Actual size of
- * CAR output may be bigger due to CAR header and block encoding data.
+ * @param {import('./types').ShardingOptions} [options]
  */
  constructor(options = {}) {
  const shardSize = options.shardSize ?? SHARD_SIZE

packages/upload-client/src/types.ts

@@ -150,7 +150,19 @@ export interface Connectable {
  connection?: ConnectionView<Service>
 }
 
-export type RequestOptions = Retryable & Abortable & Connectable
+export interface RequestOptions extends Retryable, Abortable, Connectable {}
+
+export interface ShardingOptions {
+ /**
+ * The target shard size. Actual size of CAR output may be bigger due to CAR
+ * header and block encoding data.
+ */
+ shardSize?: number
+}
+
+export interface UploadOptions extends RequestOptions, ShardingOptions {
+ onStoredShard?: (meta: CARMetadata) => void
+}
 
 export interface BlobLike {
  /**

packages/upload-client/test/index.test.js

@@ -98,6 +98,64 @@ describe('uploadFile', () => {
  assert(carCID)
  assert(dataCID)
  })
+
+ it('allows custom shard size to be set', async () => {
+ const res = {
+ status: 'upload',
+ headers: { 'x-test': 'true' },
+ url: 'http://localhost:9200',
+ }
+
+ const space = await Signer.generate()
+ const agent = await Signer.generate() // The "user" that will ask the service to accept the upload
+ const file = new Blob([await randomBytes(500_000)])
+ /** @type {import('../src/types').CARLink[]} */
+ const carCIDs = []
+
+ const proofs = await Promise.all([
+ StoreCapabilities.add.delegate({
+ issuer: space,
+ audience: agent,
+ with: space.did(),
+ expiration: Infinity,
+ }),
+ UploadCapabilities.add.delegate({
+ issuer: space,
+ audience: agent,
+ with: space.did(),
+ expiration: Infinity,
+ }),
+ ])
+
+ const service = mockService({
+ store: { add: provide(StoreCapabilities.add, () => res) },
+ upload: { add: provide(UploadCapabilities.add, () => null) },
+ })
+
+ const server = Server.create({
+ id: serviceSigner,
+ service,
+ decoder: CAR,
+ encoder: CBOR,
+ })
+ const connection = Client.connect({
+ id: serviceSigner,
+ encoder: CAR,
+ decoder: CBOR,
+ channel: server,
+ })
+ await uploadFile(
+ { issuer: agent, with: space.did(), proofs, audience: serviceSigner },
+ file,
+ {
+ connection,
+ shardSize: 400_000, // should end up with 2 CAR files
+ onStoredShard: (meta) => carCIDs.push(meta.cid),
+ }
+ )
+
+ assert.equal(carCIDs.length, 2)
+ })
 })
 
 describe('uploadDirectory', () => {
@@ -188,4 +246,62 @@ describe('uploadDirectory', () => {
  assert(carCID)
  assert(dataCID)
  })
+
+ it('allows custom shard size to be set', async () => {
+ const res = {
+ status: 'upload',
+ headers: { 'x-test': 'true' },
+ url: 'http://localhost:9200',
+ }
+
+ const space = await Signer.generate()
+ const agent = await Signer.generate() // The "user" that will ask the service to accept the upload
+ const files = [new File([await randomBytes(500_000)], '1.txt')]
+ /** @type {import('../src/types').CARLink[]} */
+ const carCIDs = []
+
+ const proofs = await Promise.all([
+ StoreCapabilities.add.delegate({
+ issuer: space,
+ audience: agent,
+ with: space.did(),
+ expiration: Infinity,
+ }),
+ UploadCapabilities.add.delegate({
+ issuer: space,
+ audience: agent,
+ with: space.did(),
+ expiration: Infinity,
+ }),
+ ])
+
+ const service = mockService({
+ store: { add: provide(StoreCapabilities.add, () => res) },
+ upload: { add: provide(UploadCapabilities.add, () => null) },
+ })
+
+ const server = Server.create({
+ id: serviceSigner,
+ service,
+ decoder: CAR,
+ encoder: CBOR,
+ })
+ const connection = Client.connect({
+ id: serviceSigner,
+ encoder: CAR,
+ decoder: CBOR,
+ channel: server,
+ })
+ await uploadDirectory(
+ { issuer: agent, with: space.did(), proofs, audience: serviceSigner },
+ files,
+ {
+ connection,
+ shardSize: 400_000, // should end up with 2 CAR files
+ onStoredShard: (meta) => carCIDs.push(meta.cid),
+ }
+ )
+
+ assert.equal(carCIDs.length, 2)
+ })
 })