Feat/hypercore11 migration

package.json

@@ -16,6 +16,10 @@
     "./schema.js": {
       "types": "./dist/schema.d.ts",
       "default": "./src/schema.js"
+    },
+    "./migration.js": {
+      "types": "./dist/migration.d.ts",
+      "default": "./src/migration.js"
     }
   },
   "type": "module",
@@ -124,6 +128,7 @@
     "@comapeo/core2.0.1": "npm:@comapeo/core@2.0.1",
     "@comapeo/core4.1.4": "npm:@comapeo/core@^4.1.4",
     "@comapeo/ipc": "^2.1.0",
+    "@digidem/types": "^2.4.1",
     "@mapeo/default-config": "^6.0.0",
     "@mapeo/mock-data": "^5.0.0",
     "@sinonjs/fake-timers": "^10.0.2",
@@ -175,23 +180,22 @@
   "dependencies": {
     "@comapeo/fallback-smp": "^1.0.0",
     "@comapeo/schema": "2.3.0",
-    "@digidem/types": "^2.3.0",
     "@fastify/error": "^3.4.1",
     "@fastify/type-provider-typebox": "^4.1.0",
     "@hyperswarm/secret-stream": "^6.6.3",
     "@mapeo/crypto": "1.0.0-alpha.10",
     "@mapeo/sqlite-indexer": "^1.0.2",
     "@sinclair/typebox": "^0.33.17",
     "@sindresorhus/merge-streams": "^4.0.0",
-    "b4a": "^1.6.3",
+    "b4a": "^1.8.0",
     "bcp-47": "^2.1.0",
     "bcp-47-normalize": "^2.3.0",
     "better-sqlite3": "^11.10.0",
     "big-sparse-array": "^1.0.3",
     "bogon": "^1.1.0",
     "comapeocat": "^1.1.0",
     "compact-encoding": "^2.12.0",
-    "corestore": "6.8.4",
+    "corestore": "^7.9.2",
     "custom-error-creator": "^1.1.1",
     "debug": "^4.3.4",
     "dot-prop": "^9.0.0",
@@ -200,10 +204,11 @@
     "ensure-error": "^4.0.0",
     "fastify": "^4.0.0",
     "fastify-plugin": "^4.5.1",
-    "hyperblobs": "2.3.0",
-    "hypercore": "10.19.0",
-    "hypercore-crypto": "3.4.2",
-    "hyperdrive": "11.5.3",
+    "hyperblobs": "^2.12.1",
+    "hypercore": "^11.30.2",
+    "hypercore-crypto": "^3.7.0",
+    "hypercore-storage": "^2.9.0",
+    "hyperdrive": "^13.3.2",
     "iso-3166": "^4.3.0",
     "json-stable-stringify": "^1.1.1",
     "magic-bytes.js": "^1.10.0",
@@ -218,9 +223,9 @@
     "protomux": "^3.4.1",
     "quickbit-universal": "^2.2.0",
     "ready-resource": "^1.2.0",
-    "sodium-universal": "^4.0.0",
+    "sodium-universal": "^5.0.1",
     "start-stop-state-machine": "^1.2.0",
-    "streamx": "^2.19.0",
+    "streamx": "^2.28.0",
     "string-timing-safe-equal": "^0.1.0",
     "styled-map-package": "^3.0.0",
     "sub-encoder": "^2.1.1",

src/blob-store/downloader.js

@@ -124,6 +124,8 @@ export class Downloader extends ReadyResource {
   /** @param {Error} err */
   #handleError = (err) => {
     if (this.#ac.signal.aborted) return
+    // Ignore request cancelled errors since they are false positives
+    if ('code' in err && err.code === 'REQUEST_CANCELLED') return
     this.emit('error', err)
     this.#ac.abort(err)
   }

src/blob-store/hyperdrive-index.js

@@ -104,11 +104,15 @@ class PretendCorestore {
     if (b4a.isBuffer(opts)) {
       opts = { publicKey: opts }
     }
-    if ('key' in opts) {
+    if ('key' in opts && opts.key !== null) {
       // @ts-ignore
       opts.publicKey = opts.key
     }
-    if ('publicKey' in opts) {
+    if ('keyPair' in opts && opts.keyPair !== null) {
+      // @ts-ignore
+      opts.publicKey = opts.keyPair.publicKey
+    }
+    if ('publicKey' in opts && opts.publicKey !== null) {
       // NB! We should always add blobIndex (Hyperbee) cores to the core manager
       // before we use them here. We would only reach the addCore path if the
       // blob core is read from the hyperbee header (before it is added to the
@@ -117,9 +121,9 @@ class PretendCorestore {
         this.#coreManager.getCoreByKey(opts.publicKey) ||
         this.#coreManager.addCore(opts.publicKey, 'blob').core
       )
-    } else if (opts.name === 'db') {
+    } else if ('name' in opts && opts.name === 'db') {
       return this.#coreManager.getWriterCore('blobIndex').core
-    } else if (opts.name.includes('blobs')) {
+    } else if ('name' in opts && opts.name.includes('blobs')) {
       return this.#coreManager.getWriterCore('blob').core
     } else {
       throw new UnsupportedCorestoreOptsError({ opts })

src/core-manager/index.js

@@ -9,15 +9,17 @@ import {
 } from '../generated/extensions.js'
 import { Logger } from '../logger.js'
 import { NAMESPACES } from '../constants.js'
-import { noop } from '../utils.js'
+import { forceBitfieldExchange, noop, patchCoreReplicator } from '../utils.js'
 import { coresTable } from '../schema/project.js'
 import * as rle from './bitfield-rle.js'
 import { CoreIndex } from './core-index.js'
 import mapObject from 'map-obj'
 import ReadyResource from 'ready-resource'
+import CorestoreStorage from 'hypercore-storage'
 import {
   InvalidProjectKeyError,
   InvalidProjectSecretKeyError,
+  UnexpectedError,
 } from '../errors.js'
 
 /** @import Hypercore from 'hypercore' */
@@ -43,12 +45,14 @@ export const kCoreManagerReplicate = Symbol('replicate core manager')
 export class CoreManager extends ReadyResource {
   #corestore
   #coreIndex
+  #storage
   /** @type {CoreRecord} */
   #creatorCoreRecord
   #queries
   #encryptionKeys
   #projectExtension
   /** @type {'opened' | 'closing' | 'closed'} */
+  // @ts-ignore
   #state = 'opened'
   #haveExtension
   #deviceId
@@ -67,7 +71,7 @@ export class CoreManager extends ReadyResource {
    * @param {Buffer} options.projectKey 32-byte public key of the project creator core
    * @param {Buffer} [options.projectSecretKey] 32-byte secret key of the project creator core
    * @param {Partial<Record<Namespace, Buffer>>} [options.encryptionKeys] Encryption keys for each namespace
-   * @param {import('hypercore').HypercoreStorage} options.storage Folder to store all hypercore data
+   * @param {string} options.storage Folder to store all hypercore data
    * @param {boolean} [options.autoDownload=true] Immediately start downloading cores - should only be set to false for tests
    * @param {Logger} [options.logger]
    */
@@ -110,11 +114,13 @@ export class CoreManager extends ReadyResource {
         .prepare(),
     }
 
+    this.#storage = new CorestoreStorage(storage)
+
     // Note: the primary key here should not be used, because we do not rely on
     // corestore for key storage (i.e. we do not get cores from corestore via a
     // name, which would derive the keypair from the primary key), but setting
     // this just in case a dependency does (e.g. hyperdrive) and we miss it.
-    this.#corestore = new Corestore(storage, { primaryKey })
+    this.#corestore = new Corestore(this.#storage, { primaryKey, unsafe: true })
     // Persistent index of core keys and namespaces in the project
     this.#coreIndex = new CoreIndex()
 
@@ -290,31 +296,34 @@ export class CoreManager extends ReadyResource {
     const existingCore = this.#coreIndex.getByCoreKey(keyPair.publicKey)
     if (existingCore) return existingCore
 
+    const encryptionKey = this.#encryptionKeys[namespace]
     const { publicKey: key, secretKey } = keyPair
     const writer = !!secretKey
+    // Pass both key and keyPair to work around corestore bug where keyPair alone
+    // doesn't result in the correct core key
     const core = this.#corestore.get({
+      key,
       keyPair,
-      encryptionKey: this.#encryptionKeys[namespace],
+      encryption: encryptionKey ? { key: encryptionKey } : undefined,
     })
     if (this.#autoDownload && namespace !== 'blob') {
       // Blob downloads are managed by BlobStore
       core.download({ start: 0, end: -1 })
     }
     // Every peer adds a listener, so could have many peers
     core.setMaxListeners(0)
-    // @ts-ignore - ensure key is defined before hypercore is ready
-    core.key = key
+
     this.#coreIndex.add({ core, key, namespace, writer })
 
     // **Hack** As soon as a peer is added, eagerly send a "want" for the entire
     // core. This ensures that the peer sends back its entire bitfield.
     // Otherwise this would only happen once we call core.download()
-    core.on('peer-add', (peer) => {
+    core.on('peer-add', (/** @type {HypercorePeer} */ peer) => {
       if (core.length === 0) return
-      // **Warning** uses internal method, but should be covered by tests
-      peer._maybeWant(0, core.length)
+      forceBitfieldExchange(core, peer)
     })
 
+    patchCoreReplicator(core)
     if (writer) {
       const sendHaves = debounce(WRITER_CORE_PREHAVES_DEBOUNCE_DELAY, () => {
         for (const peer of this.creatorCore.peers) {
@@ -345,7 +354,7 @@ export class CoreManager extends ReadyResource {
           // TODO: It would be more efficient (in terms of network traffic) to
           // send a want with start = length of previous want. Need to track
           // "last want length" sent by peer.
-          peer._maybeWant(0, core.length)
+          forceBitfieldExchange(core, peer)
         }
       })
     }
@@ -486,15 +495,45 @@ export class CoreManager extends ReadyResource {
 
     for (const { core, key } of coreRecords) {
       if (key.equals(ownWriterCore.key)) continue
-      deletionPromises.push(core.purge())
+      deletionPromises.push(purgeCore(core))
     }
 
     await Promise.all(deletionPromises)
 
     this.#queries.removeCores.run({ namespace })
+
+    // This actually clears out the data
+    await this.#storage.compact()
   }
 }
 
+/**
+ * Purge data inside a hypercore.
+ * Pending on this PR being merged into hypercore11
+ * https://github.com/holepunchto/hypercore/pull/788
+ * @param {Hypercore<Hypercore.ValueEncoding, Buffer>} core
+ */
+async function purgeCore(core) {
+  const privateCore = core.core
+  // Should never happen
+  if (!privateCore) throw new UnexpectedError('Core not initialized')
+  if (privateCore.opened === false) await privateCore.opening
+  // @ts-ignore Private methods on storage
+  const tx = privateCore.storage.write()
+  tx.deleteTreeNodeRange(0, -1)
+  tx.deleteBlockRange(0, -1)
+  // Also delete the head record (fork, length, root hash): a header claiming
+  // length > 0 with no merkle roots in storage puts the core in hypercore's
+  // repair mode on next open, in which it never sends a Synchronize and so
+  // never syncs again
+  tx.deleteHead()
+  // @ts-ignore Private methods on storage
+  privateCore.bitfield.clear(tx)
+  await tx.flush()
+  await privateCore.closeAllSessions()
+  await privateCore.close()
+}
+
 /**
  * @typedef {object} HaveMsg
  * @property {Buffer} discoveryKey

src/datastore/index.js

@@ -6,6 +6,7 @@ import { discoveryKey } from 'hypercore-crypto'
 import { NAMESPACE_SCHEMAS } from '../constants.js'
 import { createMap } from '../utils.js'
 import {
+  DataStoreClosedError,
   InvalidDocSchemaError,
   InvalidVersionIdError,
   NotFoundError,
@@ -50,6 +51,7 @@ export class DataStore extends TypedEmitter {
   #pendingAppends = new Set()
   /** @type {Record<MapeoDoc['schemaName'], Set<string>>} */
   #pendingEmits
+  #closed = false
   /** @type {(coreRecord: import('../core-manager/index.js').CoreRecord) => void} */
   #handleAddCore
 
@@ -145,6 +147,7 @@ export class DataStore extends TypedEmitter {
    * @returns {Promise<Extract<MapeoDoc, TDoc>>}
    */
   async write(doc) {
+    this.#checkClosed('write')
     // @ts-ignore
     if (!NAMESPACE_SCHEMAS[this.#namespace].includes(doc.schemaName)) {
       throw new InvalidDocSchemaError({
@@ -196,6 +199,7 @@ export class DataStore extends TypedEmitter {
 
   /** @param {Buffer} buf */
   async writeRaw(buf) {
+    this.#checkClosed('write')
     const { length } = await this.#writerCore.append(buf)
     const index = length - 1
     const coreDiscoveryKey = this.#writerCore.discoveryKey
@@ -217,13 +221,23 @@ export class DataStore extends TypedEmitter {
   }
 
   async close() {
+    if (this.#closed) return
+    this.#closed = true
     // Remove the add-core listener before closing the indexer so a late
     // 'add-core' event during teardown can't call addCore() on a closed indexer
     // (which would throw 'Cannot add core after closing').
     this.#coreManager.off('add-core', this.#handleAddCore)
     await this.#coreIndexer.close()
   }
 
+  /**
+   * Throw an error if the datastore is closed.
+   * @param {string} method
+   */
+  #checkClosed(method) {
+    if (this.#closed) throw new DataStoreClosedError({ method })
+  }
+
   /**
    * Unlink all index files. This should only be called after `close()` has resolved.
    */

src/datatype/index.js

@@ -2,6 +2,7 @@ import { validate } from '@comapeo/schema'
 import { getTableConfig } from 'drizzle-orm/sqlite-core'
 import { eq, inArray, sql } from 'drizzle-orm'
 import { randomBytes } from 'node:crypto'
+import { isDeepStrictEqual } from 'node:util'
 import { noop, mutatingDeNullify } from '../utils.js'
 import {
   DocAlreadyDeletedError,
@@ -199,6 +200,13 @@ export class DataType extends TypedEmitter {
   async [kCreateOrUpdateWithDocId](docId, value) {
     const existing = await this.getByDocId(docId).catch(nullIfNotFound)
     if (existing) {
+      const hasChanges = Object.keys(value).some(
+        // @ts-ignore Key will always be a valid property
+        (key) => !isDeepStrictEqual(existing[key], value[key])
+      )
+      if (!hasChanges) {
+        return existing
+      }
       return this.update(existing.versionId, value)
     } else {
       return this[kCreateWithDocId](docId, value, { checkExisting: false })

src/errors.js

@@ -515,6 +515,12 @@ export const InvalidDrizzleJournalError = createErrorClass({
   status: 400,
 })
 
+export const DataStoreClosedError = createErrorClass({
+  code: 'DATA_STORE_CLOSED_ERROR',
+  message: 'Unable to invoke {method}, DataStore is closed',
+  status: 503,
+})
+
 /**
  * @param {unknown} err
  * @returns {null}

src/fastify-plugins/blobs.js

@@ -113,7 +113,7 @@ async function routes(fastify, options) {
       } catch (e) {
         // This matches [how Hyperblobs checks if a blob is unavailable][0].
         // [0]: https://github.com/holepunchto/hyperblobs/blob/518088d2b828082fd70a276fa2c8848a2cf2a56b/index.js#L49
-        if (ensureError(e).message === 'Block not available') {
+        if (ensureError(e).message.includes('BLOCK_NOT_AVAILABLE')) {
           reply.code(404)
           throw new BlobNotFoundError()
         } else {