fix: fix file storage race condition (#388)

* WIP initial work

* rename Rpc to LocalPeers

* Handle deviceInfo internally, id -> deviceId

* Tests for stream error handling

* remove unnecessary constructor

* return replication stream

* Attach protomux instance to peer info

* rename and re-organize

* revert changes outside scope of PR

* WIP initial work

* Tie everything together

* rename getProjectInstance

* feat: client.listLocalPeers() & `local-peers` evt

* feat: add $sync API methods

For now this simplifies the API (because we are only supporting local
sync, not remote sync over the internet) to:

- `project.$sync.getState()`
- `project.$sync.start()`
- `project.$sync.stop()`
- Events
    - `sync-state`

It's currently not possible to stop local discovery, nor is it possible
to stop sync of the metadata namespaces (auth, config, blobIndex). The
start and stop methods stop the sync of the data and blob namespaces.

Fixes #134. Stacked on #360, #358 and #356.

* feat: Add project.$waitForInitialSync() method

Fixes Add project method to download auth + config cores #233

Rather than call this inside the `client.addProject()` method, instead I
think it is better for the API consumer to call
`project.$waitForInitialSync()` after adding a project, since this
allows the implementer to give user feedback about what is happening.

* Wait for initial sync within  addProject()

* fix: don't add core bitfield until core is ready

* feat: expose deviceId on coreManager

* fix: wait for project.ready() in waitForInitialSync

* fix: skip waitForSync in tests

* don't enable/disable namespace if not needed

* start core download when created via sparse: false

* Add debug logging

This was a big lift, but necessary to be able to debug sync issues since
temporarily adding console.log statements was too much work, and
debugging requires knowing the deviceId associated with each message.

* fix timeout

* fix: Add new cores to the indexer (!!!)

This caused a day of work: a bug from months back

* remove unnecessary log stmt

* get capabilities.getMany() to include creator

* fix invite test

* keep blob cores sparse

* optional param for LocalPeers

* re-org sync and replication

Removes old replication code attached to CoreManager
Still needs tests to be updated

* update package-lock

* chore: Add debug logging

* Add new logger to discovery + dnssd

* Get invite test working

* fix manager logger

* cleanup invite test (and make it fail :(

* fix: handle duplicate connections to LocalPeers

* fix stream close before channel open

* send invite to non-existent peer

* fixed fake timers implementation for tests

* new tests for duplicate connections

* cleanup and small fix

* Better state debug logging

* chain of invites test

* fix max listeners and add skipped test

* fix: only request a core key from one peer

Reduces the number of duplicate requests for the same keys.

* cleanup members tests with new helprs

* wait for project ready when adding

* only create 4 clients for chain of invites test

* add e2e sync tests

* add published @mapeo/mock-data

* fix: don't open cores in sparse mode

Turns out this changes how core.length etc. work, which confuses things

* fix: option to skip auto download for tests

* e2e test for stop-start sync

* fix coreManager unit tests

* fix blob store tests

* fix discovery-key event

* add coreCount to sync state

* test sync with blocked peer & fix bugs

* fix datatype unit tests

* fix blobs server unit tests

* remote peer-sync-controller unit test

This is now tested in e2e tests

* fix type issues caused by bad lockfile

* ignore debug type errors

* fixes for review comments

* move utils-new into utils

* Add debug info to test that sometimes fails

* Update package-lock.json version

* remove project.ready() (breaks things)

* wait for coreOwnership write before returning

* use file storage in tests (breaks things)

* Catch race condition in CRUD tests

* fix race condition with parallel writes

* fix tests for new createManagers syntax

* fix flakey test

This test relied on `peer.connectedAt` changing in order to distinguish
connections, but sometimes `connectedAt` was the same for both peers.
This adds a 1ms delay before making the second connection, to attempt to
stop the flakiness.

---------

Co-authored-by: Andrew Chou <andrewchou@fastmail.com>

Author: gmaclennan

src/datastore/index.js

@@ -46,7 +46,9 @@ export class DataStore extends TypedEmitter {
   #writerCore
   #coreIndexer
   /** @type {Map<string, import('p-defer').DeferredPromise<void>>} */
-  #pending = new Map()
+  #pendingIndex = new Map()
+  /** @type {Set<import('p-defer').DeferredPromise<void>['promise']>} */
+  #pendingAppends = new Set()
 
   /**
    * @param {object} opts
@@ -104,11 +106,8 @@ export class DataStore extends TypedEmitter {
    * @param {MultiCoreIndexer.Entry<'binary'>[]} entries
    */
   async #handleEntries(entries) {
-    // This is needed to avoid the batch running before the append resolves, but
-    // I think this is only necessary when using random-access-memory, and will
-    // not be an issue when the indexer is on a separate thread.
-    await new Promise((res) => setTimeout(res, 0))
     await this.#batch(entries)
+    await Promise.all(this.#pendingAppends)
     // Writes to the writerCore need to wait until the entry is indexed before
     // returning, so we check if any incoming entry has a pending promise
     for (const entry of entries) {
@@ -117,7 +116,7 @@ export class DataStore extends TypedEmitter {
         coreDiscoveryKey: discoveryKey(entry.key),
         index: entry.index,
       })
-      const pending = this.#pending.get(versionId)
+      const pending = this.#pendingIndex.get(versionId)
       if (!pending) continue
       pending.resolve()
     }
@@ -143,8 +142,16 @@ export class DataStore extends TypedEmitter {
       )
     }
     const block = encode(doc)
-
+    // The indexer batch can sometimes complete before the append below
+    // resolves, so in the batch function we await any pending appends. We can't
+    // know the versionId before the append, because docs can be written in the
+    // same tick, so we can't know their index before append resolves.
+    const deferredAppend = pDefer()
+    this.#pendingAppends.add(deferredAppend.promise)
     const { length } = await this.#writerCore.append(block)
+    deferredAppend.resolve()
+    this.#pendingAppends.delete(deferredAppend.promise)
+
     const index = length - 1
     const coreDiscoveryKey = this.#writerCore.discoveryKey
     if (!coreDiscoveryKey) {
@@ -153,7 +160,7 @@ export class DataStore extends TypedEmitter {
     const versionId = getVersionId({ coreDiscoveryKey, index })
     /** @type {import('p-defer').DeferredPromise<void>} */
     const deferred = pDefer()
-    this.#pending.set(versionId, deferred)
+    this.#pendingIndex.set(versionId, deferred)
     await deferred.promise
 
     return /** @type {Extract<MapeoDoc, TDoc>} */ (

test-e2e/manager-invite.js

@@ -10,7 +10,7 @@ import {
 } from './utils.js'
 
 test('member invite accepted', async (t) => {
-  const [creator, joiner] = await createManagers(2)
+  const [creator, joiner] = await createManagers(2, t)
   connectPeers([creator, joiner])
   await waitForPeers([creator, joiner])
 
@@ -64,7 +64,7 @@ test('member invite accepted', async (t) => {
 })
 
 test('chain of invites', async (t) => {
-  const managers = await createManagers(4)
+  const managers = await createManagers(4, t)
   const [creator, ...joiners] = managers
   connectPeers(managers)
   await waitForPeers(managers)
@@ -114,7 +114,7 @@ test('chain of invites', async (t) => {
 
 // TODO: Needs fix to inviteApi to check capabilities before sending invite
 skip("member can't invite", async (t) => {
-  const managers = await createManagers(3)
+  const managers = await createManagers(3, t)
   const [creator, member, joiner] = managers
   connectPeers(managers)
   await waitForPeers(managers)
@@ -149,7 +149,7 @@ skip("member can't invite", async (t) => {
 })
 
 test('member invite rejected', async (t) => {
-  const [creator, joiner] = await createManagers(2)
+  const [creator, joiner] = await createManagers(2, t)
   connectPeers([creator, joiner])
   await waitForPeers([creator, joiner])
 

test-e2e/members.js

@@ -17,7 +17,7 @@ import {
 } from './utils.js'
 
 test('getting yourself after creating project', async (t) => {
-  const [manager] = await createManagers(1)
+  const [manager] = await createManagers(1, t)
 
   const deviceInfo = await manager.getDeviceInfo()
   const project = await manager.getProject(await manager.createProject())
@@ -49,7 +49,7 @@ test('getting yourself after creating project', async (t) => {
 })
 
 test('getting yourself after adding project (but not yet synced)', async (t) => {
-  const [manager] = await createManagers(1)
+  const [manager] = await createManagers(1, t)
 
   const deviceInfo = await manager.getDeviceInfo()
   const project = await manager.getProject(
@@ -89,7 +89,7 @@ test('getting yourself after adding project (but not yet synced)', async (t) =>
 })
 
 test('getting invited member after invite rejected', async (t) => {
-  const managers = await createManagers(2)
+  const managers = await createManagers(2, t)
   const [invitor, invitee] = managers
   connectPeers(managers)
   await waitForPeers(managers)
@@ -120,7 +120,7 @@ test('getting invited member after invite rejected', async (t) => {
 })
 
 test('getting invited member after invite accepted', async (t) => {
-  const managers = await createManagers(2)
+  const managers = await createManagers(2, t)
   const [invitor, invitee] = managers
   connectPeers(managers)
   await waitForPeers(managers)

test-e2e/project-crud.js

@@ -1,11 +1,10 @@
 import { test } from 'brittle'
 import { randomBytes } from 'crypto'
-import { KeyManager } from '@mapeo/crypto'
 import { valueOf } from '../src/utils.js'
-import { MapeoManager } from '../src/mapeo-manager.js'
-import RAM from 'random-access-memory'
-import { stripUndef } from './utils.js'
+import { createManager, stripUndef } from './utils.js'
 import { round } from './utils.js'
+import { generate } from '@mapeo/mock-data'
+import { setTimeout as delay } from 'timers/promises'
 
 /** @satisfies {Array<import('@mapeo/schema').MapeoValue>} */
 const fixtures = [
@@ -64,34 +63,40 @@ function getUpdateFixture(value) {
   }
 }
 
+const CREATE_COUNT = 100
+
 test('CRUD operations', async (t) => {
-  const manager = new MapeoManager({
-    rootKey: KeyManager.generateRootKey(),
-    projectMigrationsFolder: new URL('../drizzle/project', import.meta.url)
-      .pathname,
-    clientMigrationsFolder: new URL('../drizzle/client', import.meta.url)
-      .pathname,
-    dbFolder: ':memory:',
-    coreStorage: () => new RAM(),
-  })
+  const manager = createManager('device0', t)
 
   for (const value of fixtures) {
     const { schemaName } = value
-    t.test(`create and read ${schemaName}`, async (st) => {
+    await t.test(`create and read ${schemaName}`, async (st) => {
       const projectId = await manager.createProject()
       const project = await manager.getProject(projectId)
-      // @ts-ignore - TS can't figure this out, but we're not testing types here so ok to ignore
-      const written = await project[schemaName].create(value)
-      const read = await project[schemaName].getByDocId(written.docId)
-      st.alike(valueOf(stripUndef(written)), value, 'expected value is written')
+      const values = []
+      const writePromises = []
+      let i = 0
+      while (i++ < CREATE_COUNT) {
+        const value = valueOf(generate(schemaName)[0])
+        values.push(value)
+        writePromises.push(
+          // @ts-ignore
+          project[schemaName].create(value)
+        )
+      }
+      const written = await Promise.all(writePromises)
+      const read = await Promise.all(
+        written.map((doc) => project[schemaName].getByDocId(doc.docId))
+      )
       st.alike(written, read, 'return create() matches return of getByDocId()')
     })
-    t.test('update', async (st) => {
+    await t.test('update', async (st) => {
       const projectId = await manager.createProject()
       const project = await manager.getProject(projectId)
       // @ts-ignore
       const written = await project[schemaName].create(value)
       const updateValue = getUpdateFixture(value)
+      await delay(1) // delay to ensure updatedAt is different to createdAt
       // @ts-ignore
       const updated = await project[schemaName].update(
         written.versionId,
@@ -112,7 +117,7 @@ test('CRUD operations', async (t) => {
       st.is(written.createdAt, updated.createdAt, 'createdAt does not change')
       st.is(written.createdBy, updated.createdBy, 'createdBy does not change')
     })
-    t.test('getMany', async (st) => {
+    await t.test('getMany', async (st) => {
       const projectId = await manager.createProject()
       const project = await manager.getProject(projectId)
       const values = new Array(5).fill(null).map(() => {

test-e2e/sync.js

@@ -20,7 +20,7 @@ const SCHEMAS_INITIAL_SYNC = ['preset', 'field']
 
 test('Create and sync data', async function (t) {
   const COUNT = 5
-  const managers = await createManagers(COUNT)
+  const managers = await createManagers(COUNT, t)
   const [invitor, ...invitees] = managers
   const disconnect = connectPeers(managers, { discovery: false })
   const projectId = await invitor.createProject()
@@ -88,7 +88,7 @@ test('start and stop sync', async function (t) {
   // Checks that both peers need to start syncing for data to sync, and that
   // $sync.stop() actually stops data syncing
   const COUNT = 2
-  const managers = await createManagers(COUNT)
+  const managers = await createManagers(COUNT, t)
   const [invitor, ...invitees] = managers
   const disconnect = connectPeers(managers, { discovery: false })
   const projectId = await invitor.createProject()
@@ -154,7 +154,7 @@ test('start and stop sync', async function (t) {
 
 test('shares cores', async function (t) {
   const COUNT = 5
-  const managers = await createManagers(COUNT)
+  const managers = await createManagers(COUNT, t)
   const [invitor, ...invitees] = managers
   connectPeers(managers, { discovery: false })
   const projectId = await invitor.createProject()
@@ -192,7 +192,7 @@ test('shares cores', async function (t) {
 
 test('no sync capabilities === no namespaces sync apart from auth', async (t) => {
   const COUNT = 3
-  const managers = await createManagers(COUNT)
+  const managers = await createManagers(COUNT, t)
   const [invitor, invitee, blocked] = managers
   const disconnect1 = connectPeers(managers, { discovery: false })
   const projectId = await invitor.createProject()

test-e2e/utils.js

@@ -9,7 +9,10 @@ import { once } from 'node:events'
 import { generate } from '@mapeo/mock-data'
 import { valueOf } from '../src/utils.js'
 import { randomInt } from 'node:crypto'
+import { temporaryDirectory } from 'tempy'
+import fsPromises from 'node:fs/promises'
 
+const FAST_TESTS = true
 const projectMigrationsFolder = new URL('../drizzle/project', import.meta.url)
   .pathname
 const clientMigrationsFolder = new URL('../drizzle/client', import.meta.url)
@@ -143,30 +146,48 @@ export async function waitForPeers(managers) {
  *
  * @template {number} T
  * @param {T} count
+ * @param {import('brittle').TestInstance} t
  * @returns {Promise<import('type-fest').ReadonlyTuple<MapeoManager, T>>}
  */
-export async function createManagers(count) {
+export async function createManagers(count, t) {
   // @ts-ignore
   return Promise.all(
     Array(count)
       .fill(null)
       .map(async (_, i) => {
         const name = 'device' + i
-        const manager = createManager(name)
+        const manager = createManager(name, t)
         await manager.setDeviceInfo({ name })
         return manager
       })
   )
 }
 
-/** @param {string} [seed] */
-export function createManager(seed) {
+/**
+ * @param {string} seed
+ * @param {import('brittle').TestInstance} t
+ */
+export function createManager(seed, t) {
+  const dbFolder = FAST_TESTS ? ':memory:' : temporaryDirectory()
+  const coreStorage = FAST_TESTS ? () => new RAM() : temporaryDirectory()
+  t.teardown(async () => {
+    if (FAST_TESTS) return
+    await Promise.all([
+      fsPromises.rm(dbFolder, { recursive: true, force: true, maxRetries: 2 }),
+      // @ts-ignore
+      fsPromises.rm(coreStorage, {
+        recursive: true,
+        force: true,
+        maxRetries: 2,
+      }),
+    ])
+  })
   return new MapeoManager({
     rootKey: getRootKey(seed),
     projectMigrationsFolder,
     clientMigrationsFolder,
-    dbFolder: ':memory:',
-    coreStorage: () => new RAM(),
+    dbFolder,
+    coreStorage,
   })
 }
 

tests/local-peers.js

@@ -13,6 +13,7 @@ import { replicate } from './helpers/local-peers.js'
 import { randomBytes } from 'node:crypto'
 import NoiseSecretStream from '@hyperswarm/secret-stream'
 import Protomux from 'protomux'
+import { setTimeout as delay } from 'timers/promises'
 
 test('Send invite and accept', async (t) => {
   t.plan(3)
@@ -83,6 +84,7 @@ test('Send invite, duplicate connections', async (t) => {
 
   const destroy1 = replicate(r1, r2, { kp1, kp2 })
   const [peers1] = await once(r1, 'peers')
+  await delay(1) // Ensure that connectedAt is different
   const destroy2 = replicate(r1, r2, { kp1, kp2 })
   const [peers2] = await once(r1, 'peers')