refactor: Make rootstore require Batching and TxnDatastore (sourcenet…

…work#940)

Relevant issue(s)
Resolves sourcenetwork#683

Description
This PR change the requirement for our supported datastore to implement both ds.Batching and ds.TxnDatastore. This means that the map datastore is no longer supported.

cli/serverdump.go

@@ -15,9 +15,9 @@ import (
 	"os"
 	"os/signal"
 
-	ds "github.com/ipfs/go-datastore"
 	"github.com/spf13/cobra"
 
+	ds "github.com/sourcenetwork/defradb/datastore"
 	badgerds "github.com/sourcenetwork/defradb/datastore/badger/v3"
 	"github.com/sourcenetwork/defradb/db"
 	"github.com/sourcenetwork/defradb/errors"
@@ -36,7 +36,7 @@ var serverDumpCmd = &cobra.Command{
 		signalCh := make(chan os.Signal, 1)
 		signal.Notify(signalCh, os.Interrupt)
 
-		var rootstore ds.Batching
+		var rootstore ds.RootStore
 		var err error
 		if datastore == badgerDatastoreName {
 			info, err := os.Stat(cfg.Datastore.Badger.Path)

cli/start.go

@@ -20,7 +20,6 @@ import (
 	"strings"
 
 	badger "github.com/dgraph-io/badger/v3"
-	ds "github.com/ipfs/go-datastore"
 	ma "github.com/multiformats/go-multiaddr"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
@@ -30,6 +29,7 @@ import (
 	httpapi "github.com/sourcenetwork/defradb/api/http"
 	"github.com/sourcenetwork/defradb/client"
 	"github.com/sourcenetwork/defradb/config"
+	ds "github.com/sourcenetwork/defradb/datastore"
 	badgerds "github.com/sourcenetwork/defradb/datastore/badger/v3"
 	"github.com/sourcenetwork/defradb/db"
 	"github.com/sourcenetwork/defradb/errors"
@@ -203,7 +203,7 @@ func (di *defraInstance) close(ctx context.Context) {
 func start(ctx context.Context) (*defraInstance, error) {
 	log.FeedbackInfo(ctx, "Starting DefraDB service...")
 
-	var rootstore ds.Batching
+	var rootstore ds.RootStore
 
 	var err error
 	if cfg.Datastore.Store == badgerDatastoreName {

client/db.go

@@ -13,7 +13,6 @@ package client
 import (
 	"context"
 
-	ds "github.com/ipfs/go-datastore"
 	blockstore "github.com/ipfs/go-ipfs-blockstore"
 
 	"github.com/sourcenetwork/defradb/datastore"
@@ -28,7 +27,7 @@ type DB interface {
 	GetCollectionBySchemaID(context.Context, string) (Collection, error)
 	GetAllCollections(ctx context.Context) ([]Collection, error)
 
-	Root() ds.Batching
+	Root() datastore.RootStore
 	Blockstore() blockstore.Blockstore
 
 	NewTxn(context.Context, bool) (datastore.Txn, error)

datastore/memory/errors.go

@@ -15,5 +15,15 @@ import "github.com/sourcenetwork/defradb/errors"
 var (
 	ErrReadOnlyTxn  = errors.New("read only transaction")
 	ErrTxnDiscarded = errors.New("transaction discarded")
-	ErrTxnConflict  = errors.New("transaction conflict")
+	ErrTxnConflict  = txnConflictError{errors.New("transaction conflict")}
 )
+
+type txnConflictError struct {
+	error
+}
+
+// custom error formatting that is non idiomatic but matches the
+// Badger transaction conflict message
+func (e txnConflictError) Error() string {
+	return "Transaction Conflict. Please retry"
+}

datastore/memory/memory.go

@@ -191,7 +191,7 @@ func (d *Datastore) Put(ctx context.Context, key ds.Key, value []byte) (err erro
 
 // Query implements ds.Query
 func (d *Datastore) Query(ctx context.Context, q dsq.Query) (dsq.Results, error) {
-	re := make([]dsq.Entry, 0, d.values.Len())
+	re := make([]dsq.Entry, 0, d.values.Height())
 	iter := d.values.Iter()
 	for iter.Next() {
 		// fast forward to last inserted version

datastore/memory/txn.go

@@ -130,7 +130,7 @@ func (t *basicTxn) Query(ctx context.Context, q dsq.Query) (dsq.Results, error)
 		return nil, ErrTxnDiscarded
 	}
 	// best effort allocation
-	re := make([]dsq.Entry, 0, t.ds.values.Len()+t.ops.Len())
+	re := make([]dsq.Entry, 0, t.ds.values.Height()+t.ops.Height())
 	iter := t.ds.values.Iter()
 	iterOps := t.ops.Iter()
 	iterOpsHasValue := iterOps.Next()
@@ -208,20 +208,18 @@ func (t *basicTxn) Commit(ctx context.Context) error {
 	if t.discarded {
 		return ErrTxnDiscarded
 	}
-	if t.readOnly {
-		return ErrReadOnlyTxn
-	}
+	defer t.Discard(ctx)
 
-	c := commit{
-		tx:  t,
-		err: make(chan error),
+	if !t.readOnly {
+		c := commit{
+			tx:  t,
+			err: make(chan error),
+		}
+		t.ds.commit <- c
+		return <-c.err
 	}
-	t.ds.commit <- c
-	e := <-c.err
 
-	t.Discard(ctx)
-
-	return e
+	return nil
 }
 
 func (t *basicTxn) checkForConflicts(ctx context.Context) error {