Fix stage corruption

index.js

@@ -33,6 +33,7 @@ module.exports = class Hyperdrive extends ReadyResource {
     this.supportsMetadata = true
     this.encryptionKey = opts.encryptionKey || null
     this.monitors = new Set()
+    this.batches = new Set()
 
     this._active = opts.active !== false
     this._openingBlobs = null
@@ -164,12 +165,16 @@ module.exports = class Hyperdrive extends ReadyResource {
   }
 
   batch() {
-    return new Hyperdrive(this.corestore, this.key, {
+    const batch = new Hyperdrive(this.corestore, this.key, {
       onwait: this._onwait,
       encryptionKey: this.encryptionKey,
       _checkout: null,
       _db: this.db.batch()
     })
+    batch.on('close', () => this.batches.delete(batch))
+    this.batches.add(batch)
+
+    return batch
   }
 
   setActive(bool) {
@@ -190,6 +195,8 @@ module.exports = class Hyperdrive extends ReadyResource {
       await this.blobs.core.close()
     }
 
+    for (const batch of this.batches) await batch.close()
+
     await this.db.close()
 
     if (!this._checkout && !this._batching) {
@@ -562,6 +569,8 @@ module.exports = class Hyperdrive extends ReadyResource {
   }
 
   createWriteStream(name, { executable = false, metadata = null, dedup = false } = {}) {
+    if (this.closing) throw new Error('Closed')
+
     const self = this
 
     let destroyed = false

test.js

@@ -1845,6 +1845,58 @@ test('dedup mode', async (t) => {
   t.is(drive.blobs.core.length, len + 1)
 })
 
+test('write after close should not corrupt drive', async (t) => {
+  const platformCorestore = new Corestore(await t.tmp(), {
+    manifestVersion: 1,
+    compat: false,
+    wait: true
+  })
+  await platformCorestore.ready()
+  t.teardown(() => platformCorestore.close())
+
+  {
+    const corestore = platformCorestore.session({ writable: true })
+    await corestore.ready()
+    t.teardown(() => corestore.close())
+
+    const drive = new Hyperdrive(corestore)
+    await drive.ready()
+    t.teardown(() => drive.close())
+
+    await drive.db.put('manifest', 'hello world')
+
+    const batch = drive.batch()
+    try {
+      for (let i = 0; i < 14; i++) {
+        const stream = batch.createWriteStream('/file' + i + '.txt')
+        const close = new Promise((resolve) => stream.on('close', resolve))
+        stream.end('hello world' + i)
+        await close
+
+        if (i === 2) await drive.close()
+      }
+      await batch.flush()
+      t.fail('batch should have errored when drive is closed')
+    } catch (err) {
+      t.pass('batch should error when drive is closed')
+    }
+    await corestore.close()
+  }
+
+  {
+    const corestore = platformCorestore.session({ writable: false })
+    await corestore.ready()
+    t.teardown(() => corestore.close())
+
+    const drive = new Hyperdrive(corestore)
+    await drive.ready()
+    t.teardown(() => drive.close())
+
+    const manifest = await drive.db.get('manifest')
+    t.is(manifest.value, 'hello world', 'should correctly read manifest')
+  }
+})
+
 async function testenv(t) {
   const { teardown } = t