pouchdb-adapter-leveldb: push replication gets stuck

Bắt đầu với _changes() của pouchdb-adapter-leveldb-core và lần theo changeStream qua ReadStreamInternal của sublevel-pouch và Iterator._next() của leveldown. Điều tra lý do binding.iterator_next() trả về một mảng rỗng đối với tệp LevelDB bị kẹt được cung cấp, thay vì tài liệu đã cập nhật. Được xem là hoàn tất khi xác định được lớp chịu trách nhiệm và định nghĩa một fix cho phép thay đổi đến được opts.onChange() và replication.

Issue

We have an Electron app that runs PouchDB with the leveldb adapter in its own Node-enabled renderer process and syncs with a remote server via db.sync() with live: true and retry: true. Occasionally we have apps that will stop syncing changes from the app to the remote server. When this happens, any changes the user makes to documents are not being replicated to the remote, although pull replication seems to be working. The failure to sync will continue for a while, persisting through reboots of the app. Sometimes things will eventually "shake loose" on their own and it will continue syncing again. The only way we have found to consistently force things to sync is to delete the user's local LevelDB files and let the app recreate them and re-sync everything from the remote.

I was able to acquire one of the LevelDB files that is in a "stuck" situation and have been using it to narrow down the source of the problem. Occasionally, whiled debugging, it will start syncing again and I have to delete the database file and start again from a fresh copy. The problem is consistently reproducible using a fresh copy of this particular file, though.

I have traced the problem to pouchdb-adapter-leveldb-core's _changes() function. When I make a small change to a document in our app (making a call to db.put() in PouchDB), that triggers the _bulk_docs method of the leveldb adapter, which writes the document change to leveldb and calls levelChanges.notify(name). That triggers the eventFunction in pouchdb-utils addListener(), setup by a previous call to _changes(). That eventFunction calls db.changes(), which results in another call to _changes(). That sets up a new changesStream stream which, when piped through the throughStream, immediately hits the flush function then the unpipe listener, instead of the throughStream's transform function. Without executing the transform function, the change never gets passed up the stack via the opts.onChange() call and replicated via http adapter.

Digging a bit deeper, I see the transform function is not being called because the stream is never receiving the data from LevelDB. The changeStream.pipe(throughStream) executes the _read() method on the ReadStreamInternal instance defined in sublevel-pouchdb:

  _read() {
    var self = this;
    /* istanbul ignore if */
    if (self._destroyed) {
      return;
    }
    /* istanbul ignore if */
    if (!self._iterator) {
      return this._waiting = true;
    }
  
    self._iterator.next(function (err, key, value) {
      if (err || (key === undefined && value === undefined)) {
        if (!err && !self._destroyed) {
          self.push(null);
        }
        return self._cleanup(err);
      }
  
  
      value = self._makeData(key, value);
      if (!self._destroyed) {
        self.push(value);
      }
    });
  }

ReadStreamInternal inherits from Readable defined in readable-stream. The callback passed to self._iterator.next() is being called with undefined err, key, and value. This causes the self.push(null) line to be executed. push() calls readableAddChunk() (defined in readable-stream/lib/_stream_readable.js), passing the null chunk that it received as an argument, which eventually calls onEofChunk(), setting the stream's state to ended.:

Readable.prototype.push = function(chunk, encoding) {
  var state = this._readableState;

  if (util.isString(chunk) && !state.objectMode) {
    encoding = encoding || state.defaultEncoding;
    if (encoding !== state.encoding) {
      chunk = new Buffer(chunk, encoding);
      encoding = '';
    }
  }

  return readableAddChunk(this, state, chunk, encoding, false);
};

Under the hood, _iterator wraps an instance of Iterator defined in leveldown/iterator.js. It inherits from AbstractIterator defined in abstract-leveldown/abstract-iterator.js. _iterator.next is defined by AbstractIterator.prototype.next, which calls self._next(), which is overridden in Iterator.prototype._next shown below:

Iterator.prototype._next = function (callback) {
  var that = this

  if (this.cache && this.cache.length) {
    process.nextTick(callback, null, this.cache.pop(), this.cache.pop())
  } else if (this.finished) {
    process.nextTick(callback)
  } else {
    binding.iterator_next(this.context, function (err, array, finished) {
      if (err) return callback(err)

      that.cache = array
      that.finished = finished
      that._next(callback)
    })
  }

  return this
}

In both the syncing and the stuck case, the first time _iterator.next() is called, the underlying iterator's cache is null and finished is false, so binding.iterator_next() is called. binding.iterator_next() is a Node native module bound function.

Properly syncing case

In the properly syncing case, binding.iterator_next() gets called with array being a tuple of key and value of the change we want to sync. It sets that.cache to the array and calls that._next(callback) (a recursive call to the Iterator.prototype._next() function). This time this.cache is the array just set so process.nextTick() is called, passing key and value as arguments along with the callback.

On the next tick, _iterator.next callback in sublevel-pouchdb's _read() function is called, receiving a value which it pushes into the read stream. In the push method, this value becomes the chunk passed into readableAddChunk().

Sync "stuck" case

In this case, the binding.iterator_next() callback is called with err = null, array = [] and finished = true. When it makes its recursive call to Iterator.prototype._next() it takes the else if branch, executing process.nextTick(callback) without any other arguments, which is what eventually calls the self._iterator.next callback with undefined key and value arguments, leading to early termination of the read stream.

Is anybody familiar enough with this code to understand why, and under what circumstances leveldown would be returning no data for a document that was just updated? It seems like there is a bug here but I'm not exactly sure in which layer of code. It seems like everything is correct up until the point we are waiting for leveldown to call its iterator_next callback, passing the value to sublevel-pouchdb but instead of passing the updated value, leveldown is passing an empty array. I realize this could very well be an issue in leveldown or in sublevel-pouch's integration with leveldown. However, Leveldown has been superseded by classic-level (#9071) and sublevel-pouch was forked from level-sublevel which was described in #7929 as an outdated project with subleveldown recommended as a replacement (which has now also been superseded by abstract-level). Is there a plan for ongoing maintenance of the leveldb adapter now that the level community seems to have moved on?

Info

• Environment: (Electron 31.3.1 with nodeIntegration enabled)
• Platform: (Node 20.15.1, Chrome 126.0.6478.185)
• Adapter: (leveldb)
• Server: (Couchbase Sync Gateway)

Reproduce

I'm not sure what it is about this particular LevelDB file that causes sync to get stuck. Unfortunately I can't share the file here.