DO NOT MERGE: debug CI flake

.github/workflows/ci.yaml

@@ -9,6 +9,9 @@ on:
   pull_request:
     branches:
       - main
+
+  workflow_dispatch:
+
 env:
   BRITTLE_TIMEOUT: 60000
 jobs:
@@ -18,7 +21,7 @@ jobs:
     strategy:
       matrix:
         node-version: [lts/*]
-        os: [ubuntu-latest, macos-latest, windows-latest]
+        os: [macos-latest, windows-latest]
     timeout-minutes: 30
     steps:
       - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4.1.1 https://github.com/actions/checkout/releases/tag/v4.1.1

index.js

@@ -359,24 +359,34 @@ module.exports = class Autobase extends ReadyResource {
   }
 
   async _close () {
+    if (this.debug) console.log('awaiting promise...')
     await Promise.resolve() // defer one tick
 
     const closing = this._advancing.catch(safetyCatch)
 
+    if (this.debug) console.log('awaiting ack timer closing...')
     if (this._ackTimer) {
       this._ackTimer.stop()
       await this._ackTimer.flush()
     }
 
+    if (this.debug) console.log('awaiting wakeup closing...')
     await this._wakeup.close()
 
+    if (this.debug) console.log('awaiting handlers closing...')
     if (this._hasClose) await this._handlers.close(this.view)
+    if (this.debug) console.log('awaiting primary closing...')
     if (this._primaryBootstrap) await this._primaryBootstrap.close()
+    if (this.debug) console.log('awaiting store closing...')
     await this.store.close()
+    if (this.debug) console.log('awaiting closing...', this.currentDrainState, this.currentApplyState)
+    if (this.debug) setTimeout(() => console.log('...', this.currentDrainState, this.currentApplyState), 100)
     await closing
+    if (this.debug) console.log('closed')
   }
 
   _onError (err) {
+    if (this.debug) console.log(err)
     if (this.closing) return
     this.close().catch(safetyCatch)
 
@@ -866,33 +876,40 @@ module.exports = class Autobase extends ReadyResource {
         this.system.requestWakeup()
       }
 
+      this.currentDrainState = 1
       const remoteAdded = await this._addRemoteHeads()
       const localNodes = this._appending === null ? null : this._addLocalHeads()
 
       if (this.closing) return
+      this.currentDrainState = 2
 
       if (remoteAdded > 0 || localNodes !== null) {
         this.updating = true
       }
 
       const u = this.linearizer.update()
+      if (this.debug) console.log('before update')
       const changed = u ? await this._applyUpdate(u) : null
+      if (this.debug) console.log('after update')
       const indexed = !!this._updatingCores
 
       if (this.closing) return
+      this.currentDrainState = 3
 
       if (this.localWriter !== null && localNodes !== null) {
         await this._flushLocal(localNodes)
       }
 
       if (this.closing) return
+      this.currentDrainState = 4
 
       const flushed = (await this._flushIndexes()) ? this.system.core.getBackingCore().flushedLength : this._systemPointer
       if (this.updating || flushed > this._systemPointer) await this._advanceBootRecord(flushed)
 
       if (indexed) await this.onindex(this)
 
       if (this.closing) return
+      this.currentDrainState = 5
 
       // force reset state in worst case
       if (this._queueViewReset && this._appending === null) {
@@ -956,6 +973,7 @@ module.exports = class Autobase extends ReadyResource {
     try {
       await this._drain()
     } catch (err) {
+      if (this.debug) console.log('DEBUG', err)
       this._onError(err)
       return
     }
@@ -1420,7 +1438,9 @@ module.exports = class Autobase extends ReadyResource {
   }
 
   async _applyUpdate (u) {
+    this.currentApplyState = 0
     await this._viewStore.flush()
+    this.currentApplyState = 1
 
     if (u.undo) this._undo(u.undo)
 
@@ -1429,6 +1449,7 @@ module.exports = class Autobase extends ReadyResource {
 
     // make sure the latest changes is reflected on the system...
     await this.system.update()
+    this.currentApplyState = 2
 
     let batch = 0
     let applyBatch = []
@@ -1462,6 +1483,7 @@ module.exports = class Autobase extends ReadyResource {
 
       return u.indexed.slice(i).concat(u.tip)
     }
+    this.currentApplyState = 3
 
     for (i = u.shared; i < u.length; i++) {
       if (this.fastForwardTo !== null && this.fastForwardTo.length > this.system.core.length && b4a.equals(this.fastForwardTo.key, this.system.core.key)) {
@@ -1512,13 +1534,16 @@ module.exports = class Autobase extends ReadyResource {
 
       if (this.system.bootstrapping) await this._bootstrap()
 
+      this.currentApplyState = 4
       if (applyBatch.length && this._hasApply === true) {
         await this._handlers.apply(applyBatch, this.view, this)
       }
+      this.currentApplyState = 5
 
       update.indexers = !!this.system.indexerUpdate
 
       await this.system.flush(await this._getViewInfo(update.indexers))
+      this.currentApplyState = 6
 
       this._applying = null
 
@@ -1532,6 +1557,7 @@ module.exports = class Autobase extends ReadyResource {
       }
 
       if (!indexed) continue
+      this.currentApplyState = 7
 
       this._shiftWriter(node.writer)
 

package.json

@@ -4,7 +4,7 @@
   "description": "",
   "main": "index.js",
   "scripts": {
-    "test": "standard && brittle test/*.js",
+    "test": "standard && brittle test/flake.js",
     "test:encrypted": "standard && brittle test/*.js --encrypt-all",
     "fuzz:generated": "brittle test/reference/fuzz/generated/*.js",
     "fuzz:main": "node test/fuzz/index.js",

test/fast-forward.js

@@ -1,4 +1,3 @@
-const os = require('os')
 const test = require('brittle')
 const tmpDir = require('test-tmp')
 const cenc = require('compact-encoding')
@@ -18,8 +17,6 @@ const {
   createBase
 } = require('./helpers')
 
-const IS_MAC_OSX = os.platform() === 'darwin'
-
 test('fast-forward - simple', async t => {
   t.plan(1)
 
@@ -251,79 +248,79 @@ test('fast-forward - multiple queues', async t => {
   }
 })
 
-if (!IS_MAC_OSX) {
-  test('fast-forward - open with no remote io', async t => {
-    const { bases, stores } = await create(2, t, {
-      apply: applyOldState,
-      fastForward: true,
-      storage: () => tmpDir(t)
-    })
-
-    const [a, b] = bases
+test('fast-forward - open with no remote io', async t => {
+  const { bases, stores } = await create(2, t, {
+    apply: applyOldState,
+    fastForward: true,
+    storage: () => tmpDir(t)
+  })
 
-    await b.ready()
+  const [a, b] = bases
 
-    for (let i = 0; i < 1000; i++) {
-      await a.append('a' + i)
-    }
+  await b.ready()
 
-    await addWriterAndSync(a, b)
-    const unreplicate = replicate([a, b])
+  for (let i = 0; i < 1000; i++) {
+    await a.append('a' + i)
+  }
 
-    const core = b.view.getBackingCore()
-    const sparse = await isSparse(core)
+  await addWriterAndSync(a, b)
+  const unreplicate = replicate([a, b])
 
-    t.ok(sparse > 0)
-    t.comment('sparse blocks: ' + sparse)
+  const core = b.view.getBackingCore()
+  const sparse = await isSparse(core)
 
-    await b.append('b1')
-    await b.append('b2')
-    await b.append('b3')
+  t.ok(sparse > 0)
+  t.comment('sparse blocks: ' + sparse)
 
-    await unreplicate()
+  await b.append('b1')
+  await b.append('b2')
+  await b.append('b3')
 
-    await a.append('a1001')
+  await unreplicate()
 
-    await b.close()
+  await a.append('a1001')
 
-    const local = a.local
-    const remote = stores[1].get({ key: local.key })
+  await b.close()
 
-    const s1 = local.replicate(true)
-    const s2 = remote.replicate(false)
+  const local = a.local
+  const remote = stores[1].get({ key: local.key })
 
-    s1.pipe(s2).pipe(s1)
+  const s1 = local.replicate(true)
+  const s2 = remote.replicate(false)
 
-    await remote.download({ end: local.length }).downloaded()
+  s1.pipe(s2).pipe(s1)
 
-    s1.destroy()
-    await new Promise(resolve => s2.on('close', resolve))
+  await remote.download({ end: local.length }).downloaded()
 
-    const b2 = await createBase(stores[1].session(), a.local.key, t, {
-      apply: applyOldState
-    })
+  s1.destroy()
+  await new Promise(resolve => s2.on('close', resolve))
 
-    await b2.ready()
-    await t.execution(b2.ready())
+  const b2 = await createBase(stores[1].session(), a.local.key, t, {
+    apply: applyOldState
+  })
 
-    async function applyOldState (batch, view, base) {
-      for (const { value } of batch) {
-        if (value.add) {
-          const key = Buffer.from(value.add, 'hex')
-          await base.addWriter(key, { indexer: value.indexer })
-          continue
-        }
+  await b2.ready()
+  await t.execution(b2.ready())
 
-        if (view) await view.append(value)
-        const core = view._source.core.session
+  b2.view.getBackingCore().session.close() // close to kill pending request
 
-        // get well distributed unique index
-        const index = (view.length * 67 + view.length * 89) % core.length
-        if (core.length) await core.get(index)
+  async function applyOldState (batch, view, base) {
+    for (const { value } of batch) {
+      if (value.add) {
+        const key = Buffer.from(value.add, 'hex')
+        await base.addWriter(key, { indexer: value.indexer })
+        continue
       }
+
+      if (view) await view.append(value)
+      const core = view._source.core.session
+
+      // get well distributed unique index
+      const index = (view.length * 67 + view.length * 89) % core.length
+      if (core.length) await core.get(index)
     }
-  })
-}
+  }
+})
 
 test('fast-forward - force reset then ff', async t => {
   t.plan(9)