VersionedStore proposal

[sritchie]

The notion of a VersionedStore would make it easy to create a distributed read-only database backed by storehaus stores that loaded themselves off of disk.

import com.twitter.util.{ Duration, Future, Timer }

/**
  * ReadableStore with some notion of versioning.
  */
trait VersionedStore[-K, +V] extends ReadableStore[K, (Long, V)] {
  def currentVersion: Option[Long]
}

/**
  * Fired when an updating versioned store is not yet loaded.
  */
class StoreNotLoadedException extends RuntimeException

/**
  * The typical not-yet-loaded versioned store. This store returns
  * exceptions for every get and multiGet.
  */
object ExceptionalStore extends VersionedStore[Any, Nothing] {
  override def get(k: Any) = Future.exception(new StoreNotLoadedException)
  override val currentVersion = None
}

class ConstantVersionedStore[-K, +V](store: ReadableStore[K, V], version: Long) extends VersionedStore[K, V] {
  override val currentVersion = Some(version)

  protected def prependVersion(f: Future[Option[V]], version: Long) =
    f.map { _.map { (version, _) } }

  override def get(k: K) = prependVersion(store.get(k), version)
  override def multiGet[K1 <: K](ks: Set[K1]): Map[K1, Future[Option[(Long, V)]]] =
    store.multiGet(ks).mapValues { prependVersion(_, version) }
}

/**
  * Versioned store implementation capable of updating itself to some
  * new version, potentially asynchronously. One use case here would a
  * store that loads its backing dataset across the network from
  * HDFS. Every time a new version appeared on HDFS the watcher
  * process would switch the backing store behind the scenes.
  */
class UpdatingVersionedStore[-K, +V](updateInterval: Duration)(watcherProcess: Option[Long] => Future[Option[VersionedStore[K, V]]])
  (implicit timer: Timer)
    extends VersionedStore[K, V] {
  protected var innerStore: Future[VersionedStore[K, V]] = Future.value(ExceptionalStore)

  timer.schedule(updateInterval) {
    innerStore = innerStore
      .join(watcherProcess(currentVersion))
      .map { case (oldStore, optNewStore) => optNewStore.getOrElse(oldStore) }
  }
  override def get(k: K) = innerStore.flatMap { _.get(k) }
  override def multiGet(ks: Set[K]) =
    FutureOps.liftFutureValues(ks, innerStore.map { _.multiGet(ks) })
}

[johnynek]

Is currentVersion well defined?

It seems like different keys might have different versions.

[sritchie]

Yes, it's well defined for a given versionedstore (which is a readablestore[K, (Long, V)]. The usual ReadableStore combinators would allow you combine multiple versioned stores.