services/syncbase/store/watchable/store.go

// Copyright 2015 The Vanadium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

// Package watchable provides a Syncbase-specific store.Store wrapper that
// provides versioned storage for specified prefixes and maintains a watchable
// log of operations performed on versioned records. This log forms the basis
// for the implementation of client-facing watch as well as the sync module's
// internal watching of store updates.
//
// LogEntry records are stored chronologically, using keys of the form
// "$log:<seq>". Sequence numbers are zero-padded to ensure that the
// lexicographic order matches the numeric order.
//
// Version number records are stored using keys of the form "$version:<key>",
// where <key> is the client-specified key.
package watchable

import (
	"fmt"
	"strings"
	"sync"
	"time"

	pubutil "v.io/v23/syncbase/util"
	"v.io/x/ref/services/syncbase/store"
)

// Options configures a Store.
type Options struct {
	// Key prefixes to version and log.
	ManagedPrefixes []string
}

// Clock is an interface to a generic clock.
type Clock interface {
	// Now returns the current time.
	Now() (time.Time, error)
}

// Wrap returns a *Store that wraps the given store.Store.
func Wrap(st store.Store, clock Clock, opts *Options) (*Store, error) {
	logStart, err := getLogStartSeq(st)
	if err != nil {
		return nil, err
	}
	seq, err := getNextLogSeq(st)
	if err != nil {
		return nil, err
	}
	return &Store{
		ist:     st,
		watcher: newWatcher(logStart),
		opts:    opts,
		seq:     seq,
		Clock:   clock,
	}, nil
}

type Store struct {
	ist     store.Store
	watcher *watcher
	opts    *Options
	mu      sync.Mutex // held during transaction commits; protects seq
	seq     uint64     // the next sequence number to be used for a new commit
	// TODO(razvanm): make the clock private. The clock is used only by the
	// addSyncgroupLogRec function from the vsync package.
	Clock Clock // used to provide write timestamps
}

var _ store.Store = (*Store)(nil)

// Close implements the store.Store interface.
func (st *Store) Close() error {
	st.watcher.close()
	return st.ist.Close()
}

// Get implements the store.StoreReader interface.
func (st *Store) Get(key, valbuf []byte) ([]byte, error) {
	if !st.managesKey(key) {
		return st.ist.Get(key, valbuf)
	}
	sn := newSnapshot(st)
	defer sn.Abort()
	return sn.Get(key, valbuf)
}

// Scan implements the store.StoreReader interface.
func (st *Store) Scan(start, limit []byte) store.Stream {
	if !st.managesRange(start, limit) {
		return st.ist.Scan(start, limit)
	}
	// TODO(sadovsky): Close snapshot once stream is finished or canceled.
	return newSnapshot(st).Scan(start, limit)
}

// Put implements the store.StoreWriter interface.
func (st *Store) Put(key, value []byte) error {
	// Use watchable.Store transaction so this op gets logged.
	return store.RunInTransaction(st, func(tx store.Transaction) error {
		return tx.Put(key, value)
	})
}

// Delete implements the store.StoreWriter interface.
func (st *Store) Delete(key []byte) error {
	// Use watchable.Store transaction so this op gets logged.
	return store.RunInTransaction(st, func(tx store.Transaction) error {
		return tx.Delete(key)
	})
}

// NewTransaction implements the store.Store interface.
func (st *Store) NewTransaction() store.Transaction {
	return newTransaction(st)
}

// NewWatchableTransaction implements the Store interface.
func (st *Store) NewWatchableTransaction() *Transaction {
	return newTransaction(st)
}

// NewSnapshot implements the store.Store interface.
func (st *Store) NewSnapshot() store.Snapshot {
	return newSnapshot(st)
}

// GetOptions returns the options configured on a watchable.Store.
// TODO(rdaoud): expose watchable store through an interface and change this
// function to be a method on the store.
func GetOptions(st store.Store) (*Options, error) {
	wst := st.(*Store)
	return wst.opts, nil
}

////////////////////////////////////////
// Internal helpers

func (st *Store) managesKey(key []byte) bool {
	ikey := string(key)
	// TODO(sadovsky): Optimize, e.g. use binary search (here and below).
	for _, p := range st.opts.ManagedPrefixes {
		if strings.HasPrefix(ikey, p) {
			return true
		}
	}
	return false
}

func (st *Store) managesRange(start, limit []byte) bool {
	istart, ilimit := string(start), limitForCompare(string(limit))
	for _, p := range st.opts.ManagedPrefixes {
		pstart, plimit := pubutil.PrefixRangeStart(p), limitForCompare(pubutil.PrefixRangeLimit(p))
		if pstart <= istart && ilimit <= plimit {
			return true
		}
		if !(plimit <= istart || ilimit <= pstart) {
			// If this happens, there's a bug in the Syncbase server implementation.
			panic(fmt.Sprintf("partial overlap: %q %q %q", p, start, limit))
		}
	}
	return false
}

func limitForCompare(limit string) string {
	if limit == "" {
		// Empty limit means no limit. Return a value always greater than a valid start.
		return "\xff"
	}
	return limit
}

func (st *Store) getSeq() uint64 {
	st.mu.Lock()
	defer st.mu.Unlock()
	return st.seq
}