Watch - high-performance replacement for subscribe()

[aboodman]

Replicache UIs are currently built using the subscribe() method.

The idea of subscribe is that it is an arbitrary function of a Replicache state that can return any JSONValue as a result. Since it is an arbitrary function there's a limit to how far it can be optimized. For example, the overwhelmingly most common subscription is just getting all entries with a particular key prefix:

rep.subscribe(async tx => {
  return await tx.scan({prefix: "todo/"}).entries().toArray();
})

We can optimize this subscription to only run when one of the accessed keys is modified (and we do currently do this). But since we can't know what the function does with that data, we are forced to run the function again each time it is invalidated, in its entirety.

This sucks since in this common case, Replicache knows how those keys changed and so it could in theory fix up the returned data surgically without re-running the scan.

We plan to support this common use case efficiently by introducing a new watch() method that looks something like:

class Replicache {
  // watch() allows developers to watch a query over the Repliache keyspace.
  // The watch is maintained incrementally, meaning that when Replicache
  // changes such that the watch is invalidated, only a small amount of work --
  // roughly proporitional to size of change -- is done to update the result.
  //
  // Specifically, in contrast to subscribe(), watch does *not* rescan the
  // keyspace on each change, nor does it do a deep compare of prev result to
  // new result to know whether a change has happened.
  watch(options?: WatchOptions): Unwatch;
}

type Entry<Key> = {
  key: Key,
  value: ReadOnlyJSONValue,
};

// Note: this is a long way of saying that the WatchOptions are the same as
// the ScanOptions but with the extension of WatchOptionsExt below. I think
// there is a refactor possible to clean this up, I can talk about that
// separately.
type WatchOptions = WatchNoIndexOptions | WatchIndexOptions;
type WatchNoIndexOptions = ScanNoIndexOptions & WatchOptionsExt<string>;
type WatchIndexOptions = ScanIndexOptions & WatchOptionsExt<[string, string]>;

type WatchOptionsExt<Key extends string | [string,string]> {
  filter?: (entry: Entry<Key>) => boolean,
  sort?: (a: Entry<Key>, b: Entry<Key>) => number,

  // Guarantees:
  // - Array items whose identities are unchanged are guaranteed to have
  //   unchanged contents. This is done specifically so that e.g., React devs
  //   can use React.memo() with these items.
  // - A single Replicache transaction results in at most 1 call to `onChange`.
  //
  // Non-guarantees:
  // - No guarantee as to identity of `result`. `result` may be reused across
  //   calls to `onChange` for performance reasons.
  // - Replicache may collapse multiple transactions into one call to
  //   `onResult`.
  onResult?: (result: Entry[]) => void,

  // Intended for use with systems like svelte, solid, mobx, etc.
  onChange?: (changes: WatchChange[]),
}

type Unwatch = () => void;

type WatchChange = {
  at: number,
  delete: number,
  insert: Entry[],
};

[KeKs0r]

I think the onRaw is quite important. There are also in react land several use cases to have an imperative API for data changes:

1. Integration with existing or other state management (in my case MOBX), without loosing the performance benefit
2. Integration with imperative modules:

• in my case a search index (via minisearch)
• a toaster (via react-toast)

3. In cases where 2 or more queries need to be joined, this mechanism will not work in order to make the resulting objects "memoizable", since the join will probably end in a new object.

For me having the access to an imperative API via the experimental watch was great not only due to performance characteristics, but also due to the flexibility of the API.

[aboodman]

Good point @KeKs0r - thanks for the feedback.

[aboodman]

Somewhere along the line we lost the fact that when watch fires it has to communicate the root hash of the database somewhere. This is needed for one of the original motivations of this bug -- keeping some external storage (e.g., a text index) up to date correctly. This can be an extra param to onResult and onChange. See #839 for the background.

[thdxr]

curious about what stage this is at

I'm currently debating between replicache and rxdb and this is the feature that it's coming down to

[aboodman]

We actually already have experimentalWatch: https://doc.replicache.dev//api/classes/Replicache. It’s deployed in current release. The new api will be a superset of this so you can start using it now.

[mashpie]

watch is very much appreciated. Below you'll find the result of some internal training for implementing replicache as vue3 store. In this case taskList is exported as reactive javascript Map(). subscribe basically triggers re-render check for all items, while we can limit that to changes only with watch - hope this will make it's way to stable api

// ./states/tasks.js
import { ref } from 'vue'
import { Replicache } from 'replicache'
import { id as newId } from '@rnvf/id-generator'

export const taskList = ref(new Map())

const rep = new Replicache({
  // eslint-disable-next-line no-undef
  licenseKey: import.meta.env.VITE_REPLICACHE_LICENSE_KEY,
  name: `tasks`,
  pushURL: '/api/tasks/push',
  pullURL: '/api/tasks/pull',
  mutators: {
    async addTask(tx, task) {
      const id = newId('task')
      await tx.put(`task/${id}`, { ...task, id, done: false })
    },
    async toggleTask(tx, task) {
      await tx.put(`task/${task.id}`, { ...task, done: !task.done })
    },
    async removeTask(tx, task) {
      await tx.del(`task/${task.id}`)
    }
  }
})

/**
 * stable but rerenders whole list (on each change)
 */
rep.subscribe(async (tx) => tx.scan({ prefix: 'task/' }).entries().toArray(), {
  onData(data) {
    // used for initial load only
    if (taskList.value.size <= 0) {
      taskList.value = new Map(data)
    }
  }
})

/**
 * experimental but rerenders only changed items
 */
rep.experimentalWatch(
  (diff) => {
    for (const { op, key, newValue } of diff) {
      if (op !== 'del') {
        taskList.value.set(key, newValue)
      } else {
        taskList.value.delete(key)
      }
    }
  },
  {
    prefix: 'task/'
    // initialValuesInFirstDiff: true // might have performance impact, too
  }
)

export const addTask = rep.mutate.addTask
export const removeTask = rep.mutate.removeTask
export const toggleTask = rep.mutate.toggleTask

only one thing left for testing: Is list = new Map(data) better than data.forEach((d) => list.set(d.id, d))?

[aboodman]

Very cool @mashpie - is there anything missing from watch for your use in Vue?

[mashpie]

Thanks @aboodman - so far I don't miss anything but time to play and setup further examples ;)