fix: blockstore operations should throw when passed an aborted signal (…

…#497)

If an aborted `AbortSignal` is passed to blockstore operations, the
operation should throw an `AbortError`.

packages/utils/src/storage.ts

@@ -62,6 +62,7 @@ export class BlockStorage implements Blocks, Startable {
    * Put a block to the underlying datastore
    */
   async put (cid: CID, block: Uint8Array, options: AbortOptions & ProgressOptions<PutBlockProgressEvents> = {}): Promise<CID> {
+    options?.signal?.throwIfAborted()
     const releaseLock = await this.lock.readLock()
 
     try {
@@ -75,6 +76,7 @@ export class BlockStorage implements Blocks, Startable {
    * Put a multiple blocks to the underlying datastore
    */
   async * putMany (blocks: AwaitIterable<{ cid: CID, block: Uint8Array }>, options: AbortOptions & ProgressOptions<PutManyBlocksProgressEvents> = {}): AsyncIterable<CID> {
+    options?.signal?.throwIfAborted()
     const releaseLock = await this.lock.readLock()
 
     try {
@@ -88,6 +90,7 @@ export class BlockStorage implements Blocks, Startable {
    * Get a block by cid
    */
   async get (cid: CID, options: GetOfflineOptions & AbortOptions & ProgressOptions<GetBlockProgressEvents> = {}): Promise<Uint8Array> {
+    options?.signal?.throwIfAborted()
     const releaseLock = await this.lock.readLock()
 
     try {
@@ -101,6 +104,7 @@ export class BlockStorage implements Blocks, Startable {
    * Get multiple blocks back from an (async) iterable of cids
    */
   async * getMany (cids: AwaitIterable<CID>, options: GetOfflineOptions & AbortOptions & ProgressOptions<GetManyBlocksProgressEvents> = {}): AsyncIterable<Pair> {
+    options?.signal?.throwIfAborted()
     const releaseLock = await this.lock.readLock()
 
     try {
@@ -114,6 +118,7 @@ export class BlockStorage implements Blocks, Startable {
    * Delete a block from the blockstore
    */
   async delete (cid: CID, options: AbortOptions & ProgressOptions<DeleteBlockProgressEvents> = {}): Promise<void> {
+    options?.signal?.throwIfAborted()
     const releaseLock = await this.lock.writeLock()
 
     try {
@@ -131,6 +136,7 @@ export class BlockStorage implements Blocks, Startable {
    * Delete multiple blocks from the blockstore
    */
   async * deleteMany (cids: AwaitIterable<CID>, options: AbortOptions & ProgressOptions<DeleteManyBlocksProgressEvents> = {}): AsyncIterable<CID> {
+    options?.signal?.throwIfAborted()
     const releaseLock = await this.lock.writeLock()
 
     try {
@@ -151,6 +157,7 @@ export class BlockStorage implements Blocks, Startable {
   }
 
   async has (cid: CID, options: AbortOptions = {}): Promise<boolean> {
+    options?.signal?.throwIfAborted()
     const releaseLock = await this.lock.readLock()
 
     try {
@@ -161,6 +168,7 @@ export class BlockStorage implements Blocks, Startable {
   }
 
   async * getAll (options: AbortOptions & ProgressOptions<GetAllBlocksProgressEvents> = {}): AsyncIterable<Pair> {
+    options?.signal?.throwIfAborted()
     const releaseLock = await this.lock.readLock()
 
     try {
@@ -171,6 +179,7 @@ export class BlockStorage implements Blocks, Startable {
   }
 
   async createSession (root: CID, options?: AbortOptions): Promise<Blockstore> {
+    options?.signal?.throwIfAborted()
     const releaseLock = await this.lock.readLock()
 
     try {

packages/utils/test/storage.spec.ts

@@ -51,6 +51,17 @@ describe('storage', () => {
     expect(retrieved).to.equalBytes(block)
   })
 
+  it('aborts getting a block from the blockstore when passed an aborted signal', async () => {
+    const { cid } = blocks[0]
+    const controller = new AbortController()
+    controller.abort()
+
+    await expect(storage.get(cid, {
+      signal: controller.signal
+    })).to.eventually.be.rejected
+      .with.property('name', 'AbortError')
+  })
+
   it('gets many blocks from the blockstore', async () => {
     const count = 5
 
@@ -69,6 +80,17 @@ describe('storage', () => {
     expect(retrieved).to.deep.equal(new Array(count).fill(0).map((_, i) => blocks[i]))
   })
 
+  it('aborts getting many blocks from the blockstore when passed an aborted signal', async () => {
+    const { cid } = blocks[0]
+    const controller = new AbortController()
+    controller.abort()
+
+    await expect(all(storage.getMany([cid], {
+      signal: controller.signal
+    }))).to.eventually.be.rejected
+      .with.property('name', 'AbortError')
+  })
+
   it('puts a block into the blockstore', async () => {
     const { cid, block } = blocks[0]
     await storage.put(cid, block)
@@ -77,6 +99,17 @@ describe('storage', () => {
     expect(retrieved).to.equalBytes(block)
   })
 
+  it('aborts putting a block into the blockstore when passed an aborted signal', async () => {
+    const { cid, block } = blocks[0]
+    const controller = new AbortController()
+    controller.abort()
+
+    await expect(storage.put(cid, block, {
+      signal: controller.signal
+    })).to.eventually.be.rejected
+      .with.property('name', 'AbortError')
+  })
+
   it('puts many blocks into the blockstore', async () => {
     const count = 5
 
@@ -90,4 +123,15 @@ describe('storage', () => {
     const retrieved = await all(blockstore.getMany(new Array(count).fill(0).map((_, i) => blocks[i].cid)))
     expect(retrieved).to.deep.equal(retrieved)
   })
+
+  it('aborts putting many blocks into the blockstore when passed an aborted signal', async () => {
+    const { cid, block } = blocks[0]
+    const controller = new AbortController()
+    controller.abort()
+
+    await expect(all(storage.putMany([{ cid, block }], {
+      signal: controller.signal
+    }))).to.eventually.be.rejected
+      .with.property('name', 'AbortError')
+  })
 })