Repository: googleforgames/open-saves
Pull request #227 (2021-04-15 23:25:12)

refactor: Make MetaDB Datastore only


<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://github.com/googleforgames/open-saves/blob/main/docs/contributing.md and developer guide https://github.com/googleforgames/open-saves/blob/main/docs/development.md
2. Please label this pull request according to what type of issue you are addressing.
3. Ensure you have added or ran the appropriate tests for your PR.
-->

**What type of PR is this?**
/kind refactor

**What this PR does / Why we need it**:
Remove the intermediate MetaDB interface and move methods in
the datastore.Driver struct to the new MetaDB struct.
We will focus on the GCP implementation and remove the redundant code.

This is per #174.

**Which issue(s) this PR fixes**:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Closes #<issue number>`, or `Closes (paste link of issue)`.
-->
Closes #

**Special notes for your reviewer**:


yuryu: That was the original intention, but it actually didn't work well as intended. We want to focus on the GCP implementation for now and expedite the development process. We could add an abstraction layer later when we need to.
loranger2k: But we could still have an interface defined at the MetaDB level, no? 
yuryu: We could, but that doesn't guarantee the interface can be implemented with another provider. Also I think this was the wrong abstraction. We should implement more primitive operations as a driver and move most of the logic out of it, but we ended up with putting everything in the driver. If we want to implement an additional backend provider, I would suggest change the implementation model. I'd like to remove the development overhead until we have a better model.
loranger2k: Sure. I suggested to leave the interface since nothing changed in any of the functions after the refactor.

Patch:
diff --git a/Makefile b/Makefile
index bc34193..336be9c 100644
--- a/Makefile
+++ b/Makefile
@@ -19,9 +19,9 @@ PROTOC = protoc
 SERVER_BIN = ${BIN_DIR}/server
 COLLECTOR_BIN = ${BIN_DIR}/collector
 OPENSAVES_GO_PROTOS = ${API_DIR}/open_saves.pb.go ${API_DIR}/open_saves_grpc.pb.go
-ALL_TARGETS = ${SERVER_BIN} ${COLLECTOR_BIN} ${OPENSAVES_GO_PROTOS} internal/pkg/metadb/mock/mock_metadb.go
+ALL_TARGETS = ${SERVER_BIN} ${COLLECTOR_BIN} ${OPENSAVES_GO_PROTOS}
 
-.PHONY: all clean test server protos swagger mock FORCE
+.PHONY: all clean test server protos swagger FORCE
 
 all: server collector
 
@@ -29,10 +29,10 @@ server: ${SERVER_BIN}
 
 collector: ${COLLECTOR_BIN}
 
-${SERVER_BIN}: cmd/server/main.go protos mock FORCE
+${SERVER_BIN}: cmd/server/main.go protos FORCE
 	go build -o $@ $<
 
-${COLLECTOR_BIN}: cmd/collector/main.go protos mock FORCE
+${COLLECTOR_BIN}: cmd/collector/main.go protos FORCE
 	go build -o $@ $<
 
 clean:
@@ -43,11 +43,6 @@ test: server
 
 protos: ${OPENSAVES_GO_PROTOS}
 
-mock: internal/pkg/metadb/mock/mock_metadb.go
-
-internal/pkg/metadb/mock/mock_metadb.go: internal/pkg/metadb/metadb.go
-	mockgen -source "$<" -destination "$@"
-
 ${OPENSAVES_GO_PROTOS}: ${API_DIR}/open_saves.proto
 	$(PROTOC) -I. \
 		--go_out=. --go_opt=paths=source_relative \
diff --git a/internal/app/collector/collector.go b/internal/app/collector/collector.go
index 074ea46..19f2f17 100644
--- a/internal/app/collector/collector.go
+++ b/internal/app/collector/collector.go
@@ -22,7 +22,6 @@ import (
 	"github.com/googleforgames/open-saves/internal/pkg/blob"
 	"github.com/googleforgames/open-saves/internal/pkg/cache"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb"
-	"github.com/googleforgames/open-saves/internal/pkg/metadb/datastore"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/api/iterator"
 )
@@ -55,13 +54,9 @@ func newCollector(ctx context.Context, cfg *Config) (*Collector, error) {
 		if err != nil {
 			return nil, err
 		}
-		datastore, err := datastore.NewDriver(ctx, cfg.Project)
+		metadb, err := metadb.NewMetaDB(ctx, cfg.Project)
 		if err != nil {
-			return nil, err
-		}
-		metadb := metadb.NewMetaDB(datastore)
-		if err := metadb.Connect(ctx); err != nil {
-			log.Fatalf("Failed to connect to the metadata server: %v", err)
+			log.Fatalf("Failed to create a MetaDB instance: %v", err)
 			return nil, err
 		}
 		redis := cache.NewRedis(cfg.Cache)
diff --git a/internal/app/server/open_saves.go b/internal/app/server/open_saves.go
index f8d8a97..d4c8b9c 100644
--- a/internal/app/server/open_saves.go
+++ b/internal/app/server/open_saves.go
@@ -25,7 +25,6 @@ import (
 	"github.com/googleforgames/open-saves/internal/pkg/blob"
 	"github.com/googleforgames/open-saves/internal/pkg/cache"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb"
-	"github.com/googleforgames/open-saves/internal/pkg/metadb/datastore"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -63,13 +62,9 @@ func newOpenSavesServer(ctx context.Context, cloud, project, bucket, cacheAddr s
 		if err != nil {
 			return nil, err
 		}
-		datastore, err := datastore.NewDriver(ctx, project)
+		metadb, err := metadb.NewMetaDB(ctx, project)
 		if err != nil {
-			return nil, err
-		}
-		metadb := metadb.NewMetaDB(datastore)
-		if err := metadb.Connect(ctx); err != nil {
-			log.Fatalf("Failed to connect to the metadata server: %v", err)
+			log.Fatalf("Failed to create a MetaDB instance: %v", err)
 			return nil, err
 		}
 		redis := cache.NewRedis(cacheAddr)
diff --git a/internal/pkg/metadb/datastore/cursor.go b/internal/pkg/metadb/cursor.go
similarity index 83%
rename from internal/pkg/metadb/datastore/cursor.go
rename to internal/pkg/metadb/cursor.go
index 1925c90..6d45e3a 100644
--- a/internal/pkg/metadb/datastore/cursor.go
+++ b/internal/pkg/metadb/cursor.go
@@ -12,16 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package datastore
+package metadb
 
 import (
 	"errors"
 
 	ds "cloud.google.com/go/datastore"
-	"github.com/googleforgames/open-saves/internal/pkg/metadb"
 )
 
-var _ metadb.BlobRefCursor = new(blobRefCursor)
+var _ BlobRefCursor = new(blobRefCursor)
 
 type blobRefCursor struct {
 	iter *ds.Iterator
@@ -31,14 +30,14 @@ func newBlobRefIterator(i *ds.Iterator) *blobRefCursor {
 	return &blobRefCursor{iter: i}
 }
 
-func (i *blobRefCursor) Next() (*metadb.BlobRef, error) {
+func (i *blobRefCursor) Next() (*BlobRef, error) {
 	if i == nil {
 		return nil, errors.New("BlobRefIterator.Next was called on nil")
 	}
 	if i.iter == nil {
 		return nil, errors.New("iterator is nil")
 	}
-	var blob metadb.BlobRef
+	var blob BlobRef
 	_, err := i.iter.Next(&blob)
 	if err != nil {
 		return nil, err
diff --git a/internal/pkg/metadb/datastore/datastore.go b/internal/pkg/metadb/datastore/datastore.go
deleted file mode 100644
index a529c5a..0000000
--- a/internal/pkg/metadb/datastore/datastore.go
+++ /dev/null
@@ -1,501 +0,0 @@
-// Copyright 2020 Google LLC
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package datastore
-
-import (
-	"context"
-	"time"
-
-	ds "cloud.google.com/go/datastore"
-	"github.com/google/uuid"
-	m "github.com/googleforgames/open-saves/internal/pkg/metadb"
-	"google.golang.org/api/iterator"
-	"google.golang.org/api/option"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
-)
-
-const (
-	storeKind          = "store"
-	recordKind         = "record"
-	blobKind           = "blob"
-	timestampPrecision = 1 * time.Microsecond
-)
-
-// Driver is an implementation of the metadb.Driver interface for Google Cloud Datastore.
-// Call NewDriver to create a new driver instance.
-type Driver struct {
-	client *ds.Client
-	// Datastore namespace for multi-tenancy
-	Namespace string
-}
-
-func (d *Driver) mutateSingleInTransaction(tx *ds.Transaction, mut *ds.Mutation) error {
-	_, err := tx.Mutate(mut)
-	if err != nil {
-		if merr, ok := err.(ds.MultiError); ok {
-			err = merr[0]
-		}
-		return datastoreErrToGRPCStatus(err)
-	}
-	return nil
-}
-
-func (d *Driver) mutateSingle(ctx context.Context, mut *ds.Mutation) error {
-	_, err := d.client.Mutate(ctx, mut)
-	if err != nil {
-		if merr, ok := err.(ds.MultiError); ok {
-			err = merr[0]
-		}
-		return datastoreErrToGRPCStatus(err)
-	}
-	return nil
-}
-
-func (d *Driver) createStoreKey(key string) *ds.Key {
-	k := ds.NameKey(storeKind, key, nil)
-	k.Namespace = d.Namespace
-	return k
-}
-
-func (d *Driver) createRecordKey(storeKey, key string) *ds.Key {
-	sk := d.createStoreKey(storeKey)
-	sk.Namespace = d.Namespace
-	rk := ds.NameKey(recordKind, key, sk)
-	rk.Namespace = d.Namespace
-	return rk
-}
-
-func (d *Driver) createBlobKey(key uuid.UUID) *ds.Key {
-	k := ds.NameKey(blobKind, key.String(), nil)
-	k.Namespace = d.Namespace
-	return k
-}
-
-func (d *Driver) newQuery(kind string) *ds.Query {
-	if d.Namespace != "" {
-		return ds.NewQuery(kind).Namespace(d.Namespace)
-	}
-	return ds.NewQuery(kind)
-}
-
-// NewDriver creates a new instance of Driver that can be used by metadb.MetaDB.
-// projectID: Google Cloud Platform project ID to use
-func NewDriver(ctx context.Context, projectID string, opts ...option.ClientOption) (*Driver, error) {
-	client, err := ds.NewClient(ctx, projectID, opts...)
-	if err != nil {
-		return nil, datastoreErrToGRPCStatus(err)
-	}
-	return &Driver{client: client}, nil
-}
-
-// Connect initiates the database connection.
-func (d *Driver) Connect(ctx context.Context) error {
-	// noop for Cloud Datastore
-	return nil
-}
-
-// Disconnect terminates the database connection.
-// Make sure to call this method to release resources (e.g. using defer).
-// The MetaDB instance will not be available after Disconnect().
-func (d *Driver) Disconnect(ctx context.Context) error {
-	return datastoreErrToGRPCStatus(d.client.Close())
-}
-
-// CreateStore creates a new store.
-func (d *Driver) CreateStore(ctx context.Context, store *m.Store) (*m.Store, error) {
-	key := d.createStoreKey(store.Key)
-	mut := ds.NewInsert(key, store)
-	if err := d.mutateSingle(ctx, mut); err != nil {
-		return nil, err
-	}
-	return store, nil
-}
-
-// GetStore fetches a store based on the key provided.
-// Returns error if the key is not found.
-func (d *Driver) GetStore(ctx context.Context, key string) (*m.Store, error) {
-	dskey := d.createStoreKey(key)
-	store := new(m.Store)
-	err := d.client.Get(ctx, dskey, store)
-	if err != nil {
-		return nil, datastoreErrToGRPCStatus(err)
-	}
-	return store, nil
-}
-
-// FindStoreByName finds and fetches a store based on the name (complete match).
-func (d *Driver) FindStoreByName(ctx context.Context, name string) (*m.Store, error) {
-	query := d.newQuery(storeKind).Filter("Name =", name)
-	iter := d.client.Run(ctx, query)
-	store := new(m.Store)
-	_, err := iter.Next(store)
-	if err != nil {
-		return nil, status.Errorf(codes.NotFound,
-			"Store (name=%s) was not found.", name)
-	}
-	return store, nil
-}
-
-// DeleteStore deletes the store with specified key.
-// Returns error if the store has any child records.
-func (d *Driver) DeleteStore(ctx context.Context, key string) error {
-	dskey := d.createStoreKey(key)
-
-	_, err := d.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
-		query := ds.NewQuery(recordKind).Transaction(tx).KeysOnly().
-			Ancestor(dskey).Limit(1).Namespace(d.Namespace)
-		iter := d.client.Run(ctx, query)
-		_, err := iter.Next(nil)
-		if err != iterator.Done {
-			return status.Errorf(codes.FailedPrecondition,
-				"DeleteStore was called for a non-empty store (%s)", key)
-		}
-		return tx.Delete(dskey)
-	})
-	if err != nil {
-		return datastoreErrToGRPCStatus(err)
-	}
-	return nil
-}
-
-// InsertRecord creates a new Record in the store specified with storeKey.
-// Returns error if there is already a record with the same key.
-func (d *Driver) InsertRecord(ctx context.Context, storeKey string, record *m.Record) (*m.Record, error) {
-	rkey := d.createRecordKey(storeKey, record.Key)
-	_, err := d.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
-		dskey := d.createStoreKey(storeKey)
-		query := ds.NewQuery(storeKind).Transaction(tx).Namespace(d.Namespace).
-			KeysOnly().Filter("__key__ = ", dskey).Limit(1)
-		iter := d.client.Run(ctx, query)
-		_, err := iter.Next(nil)
-		if err == iterator.Done {
-			return status.Errorf(codes.FailedPrecondition,
-				"InsertRecord was called with a non-existent store (%s)", storeKey)
-		}
-		mut := ds.NewInsert(rkey, record)
-		return d.mutateSingleInTransaction(tx, mut)
-	})
-	if err != nil {
-		return nil, datastoreErrToGRPCStatus(err)
-	}
-	return record, nil
-}
-
-// UpdateRecord updates the record in the store specified with storeKey.
-// Returns error if the store doesn't have a record with the key provided.
-func (d *Driver) UpdateRecord(ctx context.Context, storeKey string, key string, updater m.RecordUpdater) (*m.Record, error) {
-	var record *m.Record
-	_, err := d.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
-		rkey := d.createRecordKey(storeKey, key)
-
-		// TODO(yuryu): Consider supporting transactions in MetaDB and move
-		// this operation out of the Datastore specific code.
-		record = new(m.Record)
-		if err := tx.Get(rkey, record); err != nil {
-			return err
-		}
-
-		oldExternalBlob := record.ExternalBlob
-
-		// Update the record entry by calling the updater callback.
-		var err error
-		record, err = updater(record)
-		if err != nil {
-			return err
-		}
-
-		if oldExternalBlob != record.ExternalBlob {
-			return status.Error(codes.Internal, "UpdateRecord: ExternalBlob must not be modified in UpdateRecord")
-		}
-		// Deassociate the old blob if an external blob is associated, and a new inline blob is being added.
-		if oldExternalBlob != uuid.Nil && len(record.Blob) > 0 {
-			oldBlob, err := d.getBlobRef(ctx, tx, record.ExternalBlob)
-			if err != nil {
-				return err
-			}
-			record, err = d.markBlobRefForDeletion(ctx, tx, storeKey, record, oldBlob, uuid.Nil)
-			if err != nil {
-				return err
-			}
-		}
-
-		record.Timestamps.UpdateTimestamps(timestampPrecision)
-		return d.mutateSingleInTransaction(tx, ds.NewUpdate(rkey, record))
-	})
-	if err != nil {
-		return nil, datastoreErrToGRPCStatus(err)
-	}
-	return record, nil
-}
-
-// GetRecord fetches and returns a record with key in store storeKey.
-// Returns error if not found.
-func (d *Driver) GetRecord(ctx context.Context, storeKey, key string) (*m.Record, error) {
-	rkey := d.createRecordKey(storeKey, key)
-	record := new(m.Record)
-	if err := d.client.Get(ctx, rkey, record); err != nil {
-		return nil, datastoreErrToGRPCStatus(err)
-	}
-	return record, nil
-}
-
-// DeleteRecord deletes a record with key in store storeKey.
-// It doesn't return error even if the key is not found in the database.
-func (d *Driver) DeleteRecord(ctx context.Context, storeKey, key string) error {
-	rkey := d.createRecordKey(storeKey, key)
-	_, err := d.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
-		record := new(m.Record)
-		if err := tx.Get(rkey, record); err != nil {
-			if err == ds.ErrNoSuchEntity {
-				// Exit the transaction as DeleteRecord should ignore a not found error.
-				return nil
-			}
-			return err
-		}
-		if record.ExternalBlob != uuid.Nil {
-			blob, err := d.getBlobRef(ctx, tx, record.ExternalBlob)
-			if err == nil {
-				_, err = d.markBlobRefForDeletion(ctx, tx, storeKey, record, blob, uuid.Nil)
-				if err != nil {
-					return err
-				}
-			} else if grpc.Code(err) != codes.NotFound {
-				return err
-			}
-		}
-		return d.mutateSingleInTransaction(tx, ds.NewDelete(rkey))
-	})
-	return datastoreErrToGRPCStatus(err)
-}
-
-// TimestampPrecision returns the precision of timestamps stored in
-// Cloud Datastore. Currently it's 1 microsecond.
-// https://cloud.google.com/datastore/docs/concepts/entities#date_and_time
-func (d *Driver) TimestampPrecision() time.Duration {
-	return timestampPrecision
-}
-
-func (d *Driver) recordExists(ctx context.Context, tx *ds.Transaction, key *ds.Key) (bool, error) {
-	query := ds.NewQuery(recordKind).Namespace(d.Namespace).
-		KeysOnly().Filter("__key__ = ", key).Limit(1)
-	if tx != nil {
-		query = query.Transaction(tx)
-	}
-	iter := d.client.Run(ctx, query)
-	if _, err := iter.Next(nil); err == iterator.Done {
-		return false, nil
-	} else if err != nil {
-		return false, err
-	}
-	return true, nil
-}
-
-// InsertBlobRef implements Driver.InsertBlobRef.
-func (d *Driver) InsertBlobRef(ctx context.Context, blob *m.BlobRef) (*m.BlobRef, error) {
-	rkey := d.createRecordKey(blob.StoreKey, blob.RecordKey)
-	_, err := d.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
-		if exists, err := d.recordExists(ctx, tx, rkey); err != nil {
-			return err
-		} else if !exists {
-			return status.Error(codes.FailedPrecondition, "InsertBlob was called for a non-exitent record")
-		}
-		_, err := tx.Mutate(ds.NewInsert(d.createBlobKey(blob.Key), blob))
-		return err
-	})
-	if err != nil {
-		return nil, datastoreErrToGRPCStatus(err)
-	}
-	return blob, nil
-}
-
-// UpdateBlobRef implements Driver.UpdateBlobRef.
-func (d *Driver) UpdateBlobRef(ctx context.Context, blob *m.BlobRef) (*m.BlobRef, error) {
-	_, err := d.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
-		oldBlob, err := d.getBlobRef(ctx, tx, blob.Key)
-		if err != nil {
-			return err
-		}
-		blob.Timestamps.CreatedAt = oldBlob.Timestamps.CreatedAt
-		mut := ds.NewUpdate(d.createBlobKey(blob.Key), blob)
-		_, err = tx.Mutate(mut)
-		return err
-	})
-
-	if err != nil {
-		return nil, datastoreErrToGRPCStatus(err)
-	}
-	return blob, nil
-}
-
-func (d *Driver) getBlobRef(ctx context.Context, tx *ds.Transaction, key uuid.UUID) (*m.BlobRef, error) {
-	if key == uuid.Nil {
-		return nil, status.Error(codes.FailedPrecondition, "there is no an external blob associated")
-	}
-	blob := new(m.BlobRef)
-	if tx == nil {
-		if err := d.client.Get(ctx, d.createBlobKey(key), blob); err != nil {
-			return nil, datastoreErrToGRPCStatus(err)
-		}
-	} else {
-		if err := tx.Get(d.createBlobKey(key), blob); err != nil {
-			return nil, datastoreErrToGRPCStatus(err)
-		}
-	}
-	return blob, nil
-}
-
-// GetBlobRef implements Driver.GetCurrentBlobRef.
-func (d *Driver) GetBlobRef(ctx context.Context, key uuid.UUID) (*m.BlobRef, error) {
-	return d.getBlobRef(ctx, nil, key)
-}
-
-// GetCurrentBlobRef implements Driver.GetCurrentBlobRef.
-func (d *Driver) GetCurrentBlobRef(ctx context.Context, storeKey, recordKey string) (*m.BlobRef, error) {
-	var blob *m.BlobRef
-	_, err := d.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
-		record := new(m.Record)
-		err := tx.Get(d.createRecordKey(storeKey, recordKey), record)
-		if err != nil {
-			return err
-		}
-		blob, err = d.getBlobRef(ctx, tx, record.ExternalBlob)
-		return err
-	})
-	return blob, datastoreErrToGRPCStatus(err)
-}
-
-// PromoteBlobRefToCurrent implements Driver.PromoteBlobRefToCurrent.
-func (d *Driver) PromoteBlobRefToCurrent(ctx context.Context, blob *m.BlobRef) (*m.Record, *m.BlobRef, error) {
-	record := new(m.Record)
-	_, err := d.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
-		rkey := d.createRecordKey(blob.StoreKey, blob.RecordKey)
-		if err := tx.Get(rkey, record); err != nil {
-			return err
-		}
-		if record.ExternalBlob == uuid.Nil {
-			// Simply add the new blob if previously didn't have a blob
-			// TODO(yuryu): This should be moved out of the driver.
-			// The MetaDB Driver interface currently doesn't support transactions
-			// but we need to put this part inside the transaction.
-			record.Blob = nil
-			record.Timestamps.UpdateTimestamps(timestampPrecision)
-		} else {
-			// Mark previous blob for deletion
-			oldBlob, err := d.getBlobRef(ctx, tx, record.ExternalBlob)
-			if err != nil {
-				return err
-			}
-			record, err = d.markBlobRefForDeletion(ctx, tx, blob.StoreKey, record, oldBlob, blob.Key)
-			if err != nil {
-				return err
-			}
-		}
-		record.BlobSize = blob.Size
-		record.ExternalBlob = blob.Key
-		if err := d.mutateSingleInTransaction(tx, ds.NewUpdate(rkey, record)); err != nil {
-			return err
-		}
-
-		if blob.Status != m.BlobRefStatusReady {
-			if blob.Ready() != nil {
-				return status.Error(codes.Internal, "blob is not ready to become current")
-			}
-			if _, err := tx.Mutate(ds.NewUpdate(d.createBlobKey(blob.Key), blob)); err != nil {
-				return err
-			}
-		}
-		return nil
-	})
-	if err != nil {
-		return nil, nil, datastoreErrToGRPCStatus(err)
-	}
-	return record, blob, nil
-}
-
-// RemoveBlobFromRecord implements Driver.RemoveBlobFromRecord.
-func (d *Driver) RemoveBlobFromRecord(ctx context.Context, storeKey string, recordKey string) (*m.Record, *m.BlobRef, error) {
-	rkey := d.createRecordKey(storeKey, recordKey)
-	blob := new(m.BlobRef)
-	record := new(m.Record)
-	_, err := d.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
-		err := tx.Get(rkey, record)
-		if err != nil {
-			return err
-		}
-
-		record.BlobSize = 0
-		if record.ExternalBlob == uuid.Nil && len(record.Blob) > 0 {
-			// Delete the inline blob
-			record.Blob = nil
-			record.Timestamps.UpdateTimestamps(timestampPrecision)
-			return d.mutateSingleInTransaction(tx, ds.NewUpdate(rkey, record))
-		}
-
-		// Otherwise, clear the association
-		blob, err = d.getBlobRef(ctx, tx, record.ExternalBlob)
-		if err != nil {
-			return err
-		}
-		record, err = d.markBlobRefForDeletion(ctx, tx, storeKey, record, blob, uuid.Nil)
-		if err != nil {
-			return err
-		}
-		return d.mutateSingleInTransaction(tx, ds.NewUpdate(rkey, record))
-	})
-	if err != nil {
-		return nil, nil, datastoreErrToGRPCStatus(err)
-	}
-	return record, blob, nil
-}
-
-// Returns a modified Record and the caller must commit the change.
-func (d *Driver) markBlobRefForDeletion(_ context.Context, tx *ds.Transaction,
-	storeKey string, record *m.Record, blob *m.BlobRef, newBlobKey uuid.UUID) (*m.Record, error) {
-	if record.ExternalBlob == uuid.Nil {
-		return nil, status.Error(codes.FailedPrecondition, "the record doesn't have an external blob associated")
-	}
-	record.ExternalBlob = newBlobKey
-	record.Timestamps.UpdateTimestamps(timestampPrecision)
-	if blob.MarkForDeletion() != nil && blob.Fail() != nil {
-		return nil, status.Errorf(codes.Internal, "failed to transition the blob state for deletion: current = %v", blob.Status)
-	}
-	_, err := tx.Mutate(ds.NewUpdate(d.createBlobKey(blob.Key), blob))
-	return record, err
-}
-
-// DeleteBlobRef implements Driver.DeleteBlobRef.
-func (d *Driver) DeleteBlobRef(ctx context.Context, key uuid.UUID) error {
-	_, err := d.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
-		blob, err := d.getBlobRef(ctx, tx, key)
-		if err != nil {
-			return err
-		}
-		if blob.Status == m.BlobRefStatusReady {
-			return status.Error(codes.FailedPrecondition, "blob is currently marked as ready. mark it for deletion first")
-		}
-		return tx.Delete(d.createBlobKey(key))
-	})
-	return datastoreErrToGRPCStatus(err)
-}
-
-// ListBlobRefsByStatus implements Driver.ListBlobRefsByStatus.
-func (d *Driver) ListBlobRefsByStatus(ctx context.Context, status m.BlobRefStatus, olderThan time.Time) (m.BlobRefCursor, error) {
-	query := d.newQuery(blobKind).Filter("Status = ", int(status)).
-		Filter("Timestamps.UpdatedAt <", olderThan).Namespace(d.Namespace)
-	iter := &blobRefCursor{d.client.Run(ctx, query)}
-	return iter, nil
-}
diff --git a/internal/pkg/metadb/datastore/error.go b/internal/pkg/metadb/error.go
similarity index 98%
rename from internal/pkg/metadb/datastore/error.go
rename to internal/pkg/metadb/error.go
index 932935c..d9b6b41 100644
--- a/internal/pkg/metadb/datastore/error.go
+++ b/internal/pkg/metadb/error.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package datastore
+package metadb
 
 import (
 	ds "cloud.google.com/go/datastore"
diff --git a/internal/pkg/metadb/metadb.go b/internal/pkg/metadb/metadb.go
index f978933..3fd5c5e 100644
--- a/internal/pkg/metadb/metadb.go
+++ b/internal/pkg/metadb/metadb.go
@@ -18,41 +18,23 @@ import (
 	"context"
 	"time"
 
+	ds "cloud.google.com/go/datastore"
 	"github.com/google/uuid"
+	"google.golang.org/api/iterator"
+	"google.golang.org/api/option"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
-// Driver interface defines common operations for the metadata server.
-// MetaDB uses this interface to perform actual operations on the backend service.
-type Driver interface {
-	Connect(ctx context.Context) error
-	Disconnect(ctx context.Context) error
-
-	CreateStore(ctx context.Context, store *Store) (*Store, error)
-	GetStore(ctx context.Context, key string) (*Store, error)
-	FindStoreByName(ctx context.Context, name string) (*Store, error)
-	DeleteStore(ctx context.Context, key string) error
-
-	InsertRecord(ctx context.Context, storeKey string, record *Record) (*Record, error)
-	UpdateRecord(ctx context.Context, storeKey string, key string, updater RecordUpdater) (*Record, error)
-	GetRecord(ctx context.Context, storeKey, key string) (*Record, error)
-	DeleteRecord(ctx context.Context, storeKey, key string) error
-
-	InsertBlobRef(ctx context.Context, blob *BlobRef) (*BlobRef, error)
-	UpdateBlobRef(ctx context.Context, blob *BlobRef) (*BlobRef, error)
-	GetBlobRef(ctx context.Context, key uuid.UUID) (*BlobRef, error)
-	GetCurrentBlobRef(ctx context.Context, storeKey, recordKey string) (*BlobRef, error)
-	PromoteBlobRefToCurrent(ctx context.Context, blob *BlobRef) (*Record, *BlobRef, error)
-	RemoveBlobFromRecord(ctx context.Context, storeKey string, recordKey string) (*Record, *BlobRef, error)
-	DeleteBlobRef(ctx context.Context, key uuid.UUID) error
-	ListBlobRefsByStatus(ctx context.Context, status BlobRefStatus, olderThan time.Time) (BlobRefCursor, error)
-
-	TimestampPrecision() time.Duration
-}
+const (
+	storeKind          = "store"
+	recordKind         = "record"
+	blobKind           = "blob"
+	timestampPrecision = 1 * time.Microsecond
+)
 
 // MetaDB is a metadata database manager of Open Saves.
-// It performs operations through the Driver interface.
 // The methods return gRPC error codes. Here are some common error codes
 // returned. For additional details, please look at the method help.
 // Common errors:
@@ -61,58 +43,208 @@ type Driver interface {
 //	- InvalidArgument: key or value provided is not valid
 //	- Internal: internal unrecoverable error
 type MetaDB struct {
-	driver Driver
+	// Datastore namespace for multi-tenancy
+	Namespace string
+
+	client *ds.Client
 }
 
 // RecordUpdater is a callback function for record updates.
 // Returning a non-nil error aborts the transaction.
 type RecordUpdater func(record *Record) (*Record, error)
 
-// NewMetaDB creates a new MetaDB instance with an initialized database Driver.
-func NewMetaDB(driver Driver) *MetaDB {
-	return &MetaDB{driver: driver}
+// NewMetaDB creates a new MetaDB instance with an initialized database client.
+func NewMetaDB(ctx context.Context, projectID string, opts ...option.ClientOption) (*MetaDB, error) {
+	client, err := ds.NewClient(ctx, projectID, opts...)
+	if err != nil {
+		return nil, datastoreErrToGRPCStatus(err)
+	}
+	return &MetaDB{client: client}, nil
+}
+
+func (m *MetaDB) newQuery(kind string) *ds.Query {
+	if m.Namespace != "" {
+		return ds.NewQuery(kind).Namespace(m.Namespace)
+	}
+	return ds.NewQuery(kind)
+}
+
+func (m *MetaDB) createStoreKey(key string) *ds.Key {
+	k := ds.NameKey(storeKind, key, nil)
+	k.Namespace = m.Namespace
+	return k
+}
+
+func (m *MetaDB) createRecordKey(storeKey, key string) *ds.Key {
+	sk := m.createStoreKey(storeKey)
+	sk.Namespace = m.Namespace
+	rk := ds.NameKey(recordKind, key, sk)
+	rk.Namespace = m.Namespace
+	return rk
+}
+
+func (d *MetaDB) createBlobKey(key uuid.UUID) *ds.Key {
+	k := ds.NameKey(blobKind, key.String(), nil)
+	k.Namespace = d.Namespace
+	return k
+}
+
+func (m *MetaDB) getBlobRef(ctx context.Context, tx *ds.Transaction, key uuid.UUID) (*BlobRef, error) {
+	if key == uuid.Nil {
+		return nil, status.Error(codes.FailedPrecondition, "there is no an external blob associated")
+	}
+	blob := new(BlobRef)
+	if tx == nil {
+		if err := m.client.Get(ctx, m.createBlobKey(key), blob); err != nil {
+			return nil, datastoreErrToGRPCStatus(err)
+		}
+	} else {
+		if err := tx.Get(m.createBlobKey(key), blob); err != nil {
+			return nil, datastoreErrToGRPCStatus(err)
+		}
+	}
+	return blob, nil
+}
+
+// Returns a modified Record and the caller must commit the change.
+func (m *MetaDB) markBlobRefForDeletion(_ context.Context, tx *ds.Transaction,
+	storeKey string, record *Record, blob *BlobRef, newBlobKey uuid.UUID) (*Record, error) {
+	if record.ExternalBlob == uuid.Nil {
+		return nil, status.Error(codes.FailedPrecondition, "the record doesn't have an external blob associated")
+	}
+	record.ExternalBlob = newBlobKey
+	record.Timestamps.UpdateTimestamps(timestampPrecision)
+	if blob.MarkForDeletion() != nil && blob.Fail() != nil {
+		return nil, status.Errorf(codes.Internal, "failed to transition the blob state for deletion: current = %v", blob.Status)
+	}
+	_, err := tx.Mutate(ds.NewUpdate(m.createBlobKey(blob.Key), blob))
+	return record, err
+}
+
+func (m *MetaDB) mutateSingleInTransaction(tx *ds.Transaction, mut *ds.Mutation) error {
+	_, err := tx.Mutate(mut)
+	if err != nil {
+		if merr, ok := err.(ds.MultiError); ok {
+			err = merr[0]
+		}
+		return datastoreErrToGRPCStatus(err)
+	}
+	return nil
 }
 
-// Connect initiates the database connection.
-func (m *MetaDB) Connect(ctx context.Context) error {
-	return m.driver.Connect(ctx)
+func (m *MetaDB) mutateSingle(ctx context.Context, mut *ds.Mutation) error {
+	_, err := m.client.Mutate(ctx, mut)
+	if err != nil {
+		if merr, ok := err.(ds.MultiError); ok {
+			err = merr[0]
+		}
+		return datastoreErrToGRPCStatus(err)
+	}
+	return nil
+}
+
+func (m *MetaDB) recordExists(ctx context.Context, tx *ds.Transaction, key *ds.Key) (bool, error) {
+	query := ds.NewQuery(recordKind).Namespace(m.Namespace).
+		KeysOnly().Filter("__key__ = ", key).Limit(1)
+	if tx != nil {
+		query = query.Transaction(tx)
+	}
+	iter := m.client.Run(ctx, query)
+	if _, err := iter.Next(nil); err == iterator.Done {
+		return false, nil
+	} else if err != nil {
+		return false, err
+	}
+	return true, nil
 }
 
 // Disconnect terminates the database connection.
 // Make sure to call this method to release resources (e.g. using defer).
 // The MetaDB instance will not be available after Disconnect().
 func (m *MetaDB) Disconnect(ctx context.Context) error {
-	return m.driver.Disconnect(ctx)
+	return datastoreErrToGRPCStatus(m.client.Close())
 }
 
 // CreateStore creates a new store.
 func (m *MetaDB) CreateStore(ctx context.Context, store *Store) (*Store, error) {
-	store.Timestamps.NewTimestamps(m.driver.TimestampPrecision())
-	return m.driver.CreateStore(ctx, store)
+	store.Timestamps.NewTimestamps(timestampPrecision)
+	key := m.createStoreKey(store.Key)
+	mut := ds.NewInsert(key, store)
+	if err := m.mutateSingle(ctx, mut); err != nil {
+		return nil, err
+	}
+	return store, nil
 }
 
 // GetStore fetches a store based on the key provided.
 // Returns error if the key is not found.
 func (m *MetaDB) GetStore(ctx context.Context, key string) (*Store, error) {
-	return m.driver.GetStore(ctx, key)
+	dskey := m.createStoreKey(key)
+	store := new(Store)
+	err := m.client.Get(ctx, dskey, store)
+	if err != nil {
+		return nil, datastoreErrToGRPCStatus(err)
+	}
+	return store, nil
 }
 
 // FindStoreByName finds and fetch a store based on the name (complete match).
 func (m *MetaDB) FindStoreByName(ctx context.Context, name string) (*Store, error) {
-	return m.driver.FindStoreByName(ctx, name)
+	query := m.newQuery(storeKind).Filter("Name =", name)
+	iter := m.client.Run(ctx, query)
+	store := new(Store)
+	_, err := iter.Next(store)
+	if err != nil {
+		return nil, status.Errorf(codes.NotFound,
+			"Store (name=%s) was not found.", name)
+	}
+	return store, nil
 }
 
 // DeleteStore deletes the store with specified key.
 // Returns error if the store has any child records.
 func (m *MetaDB) DeleteStore(ctx context.Context, key string) error {
-	return m.driver.DeleteStore(ctx, key)
+	dskey := m.createStoreKey(key)
+
+	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
+		query := ds.NewQuery(recordKind).Transaction(tx).KeysOnly().
+			Ancestor(dskey).Limit(1).Namespace(m.Namespace)
+		iter := m.client.Run(ctx, query)
+		_, err := iter.Next(nil)
+		if err != iterator.Done {
+			return status.Errorf(codes.FailedPrecondition,
+				"DeleteStore was called for a non-empty store (%s)", key)
+		}
+		return tx.Delete(dskey)
+	})
+	if err != nil {
+		return datastoreErrToGRPCStatus(err)
+	}
+	return nil
 }
 
 // InsertRecord creates a new Record in the store specified with storeKey.
 // Returns error if there is already a record with the same key.
 func (m *MetaDB) InsertRecord(ctx context.Context, storeKey string, record *Record) (*Record, error) {
-	record.Timestamps.NewTimestamps(m.driver.TimestampPrecision())
-	return m.driver.InsertRecord(ctx, storeKey, record)
+	record.Timestamps.NewTimestamps(timestampPrecision)
+	rkey := m.createRecordKey(storeKey, record.Key)
+	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
+		dskey := m.createStoreKey(storeKey)
+		query := ds.NewQuery(storeKind).Transaction(tx).Namespace(m.Namespace).
+			KeysOnly().Filter("__key__ = ", dskey).Limit(1)
+		iter := m.client.Run(ctx, query)
+		_, err := iter.Next(nil)
+		if err == iterator.Done {
+			return status.Errorf(codes.FailedPrecondition,
+				"InsertRecord was called with a non-existent store (%s)", storeKey)
+		}
+		mut := ds.NewInsert(rkey, record)
+		return m.mutateSingleInTransaction(tx, mut)
+	})
+	if err != nil {
+		return nil, datastoreErrToGRPCStatus(err)
+	}
+	return record, nil
 }
 
 // UpdateRecord updates the record in the store specified with storeKey.
@@ -123,39 +255,134 @@ func (m *MetaDB) UpdateRecord(ctx context.Context, storeKey string, key string,
 	if updater == nil {
 		return nil, grpc.Errorf(codes.Internal, "updater cannot be nil")
 	}
-	return m.driver.UpdateRecord(ctx, storeKey, key, updater)
+	var record *Record
+	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
+		rkey := m.createRecordKey(storeKey, key)
+
+		// TODO(yuryu): Consider supporting transactions in MetaDB and move
+		// this operation out of the Datastore specific code.
+		record = new(Record)
+		if err := tx.Get(rkey, record); err != nil {
+			return err
+		}
+
+		oldExternalBlob := record.ExternalBlob
+
+		// Update the record entry by calling the updater callback.
+		var err error
+		record, err = updater(record)
+		if err != nil {
+			return err
+		}
+
+		if oldExternalBlob != record.ExternalBlob {
+			return status.Error(codes.Internal, "UpdateRecord: ExternalBlob must not be modified in UpdateRecord")
+		}
+		// Deassociate the old blob if an external blob is associated, and a new inline blob is being added.
+		if oldExternalBlob != uuid.Nil && len(record.Blob) > 0 {
+			oldBlob, err := m.getBlobRef(ctx, tx, record.ExternalBlob)
+			if err != nil {
+				return err
+			}
+			record, err = m.markBlobRefForDeletion(ctx, tx, storeKey, record, oldBlob, uuid.Nil)
+			if err != nil {
+				return err
+			}
+		}
+
+		record.Timestamps.UpdateTimestamps(timestampPrecision)
+		return m.mutateSingleInTransaction(tx, ds.NewUpdate(rkey, record))
+	})
+	if err != nil {
+		return nil, datastoreErrToGRPCStatus(err)
+	}
+	return record, nil
 }
 
 // GetRecord fetches and returns a record with key in store storeKey.
 // Returns error if not found.
 func (m *MetaDB) GetRecord(ctx context.Context, storeKey, key string) (*Record, error) {
-	return m.driver.GetRecord(ctx, storeKey, key)
+	rkey := m.createRecordKey(storeKey, key)
+	record := new(Record)
+	if err := m.client.Get(ctx, rkey, record); err != nil {
+		return nil, datastoreErrToGRPCStatus(err)
+	}
+	return record, nil
 }
 
 // DeleteRecord deletes a record with key in store storeKey.
 // It doesn't return error even if the key is not found in the database.
 func (m *MetaDB) DeleteRecord(ctx context.Context, storeKey, key string) error {
-	return m.driver.DeleteRecord(ctx, storeKey, key)
+	rkey := m.createRecordKey(storeKey, key)
+	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
+		record := new(Record)
+		if err := tx.Get(rkey, record); err != nil {
+			if err == ds.ErrNoSuchEntity {
+				// Exit the transaction as DeleteRecord should ignore a not found error.
+				return nil
+			}
+			return err
+		}
+		if record.ExternalBlob != uuid.Nil {
+			blob, err := m.getBlobRef(ctx, tx, record.ExternalBlob)
+			if err == nil {
+				_, err = m.markBlobRefForDeletion(ctx, tx, storeKey, record, blob, uuid.Nil)
+				if err != nil {
+					return err
+				}
+			} else if grpc.Code(err) != codes.NotFound {
+				return err
+			}
+		}
+		return m.mutateSingleInTransaction(tx, ds.NewDelete(rkey))
+	})
+	return datastoreErrToGRPCStatus(err)
 }
 
 // InsertBlobRef inserts a new BlobRef object to the datastore.
 func (m *MetaDB) InsertBlobRef(ctx context.Context, blob *BlobRef) (*BlobRef, error) {
-	blob.Timestamps.NewTimestamps(m.driver.TimestampPrecision())
-	return m.driver.InsertBlobRef(ctx, blob)
+	blob.Timestamps.NewTimestamps(timestampPrecision)
+	rkey := m.createRecordKey(blob.StoreKey, blob.RecordKey)
+	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
+		if exists, err := m.recordExists(ctx, tx, rkey); err != nil {
+			return err
+		} else if !exists {
+			return status.Error(codes.FailedPrecondition, "InsertBlob was called for a non-exitent record")
+		}
+		_, err := tx.Mutate(ds.NewInsert(m.createBlobKey(blob.Key), blob))
+		return err
+	})
+	if err != nil {
+		return nil, datastoreErrToGRPCStatus(err)
+	}
+	return blob, nil
 }
 
 // UpdateBlobRef updates a BlobRef object with the new property values.
 // Returns a NotFound error if the key is not found.
 func (m *MetaDB) UpdateBlobRef(ctx context.Context, blob *BlobRef) (*BlobRef, error) {
-	blob.Timestamps.UpdateTimestamps(m.driver.TimestampPrecision())
-	return m.driver.UpdateBlobRef(ctx, blob)
+	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
+		oldBlob, err := m.getBlobRef(ctx, tx, blob.Key)
+		if err != nil {
+			return err
+		}
+		blob.Timestamps.CreatedAt = oldBlob.Timestamps.CreatedAt
+		mut := ds.NewUpdate(m.createBlobKey(blob.Key), blob)
+		_, err = tx.Mutate(mut)
+		return err
+	})
+
+	if err != nil {
+		return nil, datastoreErrToGRPCStatus(err)
+	}
+	return blob, nil
 }
 
 // GetBlobRef returns a BlobRef object specified by the key.
 // Returns errors:
 //	- NotFound: the object is not found.
 func (m *MetaDB) GetBlobRef(ctx context.Context, key uuid.UUID) (*BlobRef, error) {
-	return m.driver.GetBlobRef(ctx, key)
+	return m.getBlobRef(ctx, nil, key)
 }
 
 // GetCurrentBlobRef gets a BlobRef object associated with a record.
@@ -163,7 +390,17 @@ func (m *MetaDB) GetBlobRef(ctx context.Context, key uuid.UUID) (*BlobRef, error
 // 	- NotFound: the record is not found.
 // 	- FailedPrecondition: the record doesn't have a blob.
 func (m *MetaDB) GetCurrentBlobRef(ctx context.Context, storeKey, recordKey string) (*BlobRef, error) {
-	return m.driver.GetCurrentBlobRef(ctx, storeKey, recordKey)
+	var blob *BlobRef
+	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
+		record := new(Record)
+		err := tx.Get(m.createRecordKey(storeKey, recordKey), record)
+		if err != nil {
+			return err
+		}
+		blob, err = m.getBlobRef(ctx, tx, record.ExternalBlob)
+		return err
+	})
+	return blob, datastoreErrToGRPCStatus(err)
 }
 
 // PromoteBlobRefToCurrent promotes the provided BlobRef object as a current
@@ -172,7 +409,47 @@ func (m *MetaDB) GetCurrentBlobRef(ctx context.Context, storeKey, recordKey stri
 //	- NotFound: the specified record or the blobref was not found
 //  - Internal: BlobRef status transition error
 func (m *MetaDB) PromoteBlobRefToCurrent(ctx context.Context, blob *BlobRef) (*Record, *BlobRef, error) {
-	return m.driver.PromoteBlobRefToCurrent(ctx, blob)
+	record := new(Record)
+	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
+		rkey := m.createRecordKey(blob.StoreKey, blob.RecordKey)
+		if err := tx.Get(rkey, record); err != nil {
+			return err
+		}
+		if record.ExternalBlob == uuid.Nil {
+			// Simply add the new blob if previously didn't have a blob
+			record.Blob = nil
+			record.Timestamps.UpdateTimestamps(timestampPrecision)
+		} else {
+			// Mark previous blob for deletion
+			oldBlob, err := m.getBlobRef(ctx, tx, record.ExternalBlob)
+			if err != nil {
+				return err
+			}
+			record, err = m.markBlobRefForDeletion(ctx, tx, blob.StoreKey, record, oldBlob, blob.Key)
+			if err != nil {
+				return err
+			}
+		}
+		record.BlobSize = blob.Size
+		record.ExternalBlob = blob.Key
+		if err := m.mutateSingleInTransaction(tx, ds.NewUpdate(rkey, record)); err != nil {
+			return err
+		}
+
+		if blob.Status != BlobRefStatusReady {
+			if blob.Ready() != nil {
+				return status.Error(codes.Internal, "blob is not ready to become current")
+			}
+			if _, err := tx.Mutate(ds.NewUpdate(m.createBlobKey(blob.Key), blob)); err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, nil, datastoreErrToGRPCStatus(err)
+	}
+	return record, blob, nil
 }
 
 // RemoveBlobFromRecord removes the ExternalBlob from the record specified by
@@ -183,7 +460,38 @@ func (m *MetaDB) PromoteBlobRefToCurrent(ctx context.Context, blob *BlobRef) (*R
 //	- FailedPrecondition: the record doesn't have an external blob
 //  - Internal: BlobRef status transition error
 func (m *MetaDB) RemoveBlobFromRecord(ctx context.Context, storeKey string, recordKey string) (*Record, *BlobRef, error) {
-	return m.driver.RemoveBlobFromRecord(ctx, storeKey, recordKey)
+	rkey := m.createRecordKey(storeKey, recordKey)
+	blob := new(BlobRef)
+	record := new(Record)
+	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
+		err := tx.Get(rkey, record)
+		if err != nil {
+			return err
+		}
+
+		record.BlobSize = 0
+		if record.ExternalBlob == uuid.Nil && len(record.Blob) > 0 {
+			// Delete the inline blob
+			record.Blob = nil
+			record.Timestamps.UpdateTimestamps(timestampPrecision)
+			return m.mutateSingleInTransaction(tx, ds.NewUpdate(rkey, record))
+		}
+
+		// Otherwise, clear the association
+		blob, err = m.getBlobRef(ctx, tx, record.ExternalBlob)
+		if err != nil {
+			return err
+		}
+		record, err = m.markBlobRefForDeletion(ctx, tx, storeKey, record, blob, uuid.Nil)
+		if err != nil {
+			return err
+		}
+		return m.mutateSingleInTransaction(tx, ds.NewUpdate(rkey, record))
+	})
+	if err != nil {
+		return nil, nil, datastoreErrToGRPCStatus(err)
+	}
+	return record, blob, nil
 }
 
 // DeleteBlobRef deletes the BlobRef object from the database.
@@ -191,11 +499,24 @@ func (m *MetaDB) RemoveBlobFromRecord(ctx context.Context, storeKey string, reco
 //	- NotFound: the blobref object is not found
 //	- FailedPrecondition: the blobref status is Ready and can't be deleted
 func (m *MetaDB) DeleteBlobRef(ctx context.Context, key uuid.UUID) error {
-	return m.driver.DeleteBlobRef(ctx, key)
+	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
+		blob, err := m.getBlobRef(ctx, tx, key)
+		if err != nil {
+			return err
+		}
+		if blob.Status == BlobRefStatusReady {
+			return status.Error(codes.FailedPrecondition, "blob is currently marked as ready. mark it for deletion first")
+		}
+		return tx.Delete(m.createBlobKey(key))
+	})
+	return datastoreErrToGRPCStatus(err)
 }
 
 // ListBlobRefsByStatus returns a cursor that iterates over BlobRefs
 // where Status = status and UpdatedAt < olderThan.
 func (m *MetaDB) ListBlobRefsByStatus(ctx context.Context, status BlobRefStatus, olderThan time.Time) (BlobRefCursor, error) {
-	return m.driver.ListBlobRefsByStatus(ctx, status, olderThan)
+	query := m.newQuery(blobKind).Filter("Status = ", int(status)).
+		Filter("Timestamps.UpdatedAt <", olderThan)
+	iter := &blobRefCursor{m.client.Run(ctx, query)}
+	return iter, nil
 }


Test patch:
diff --git a/internal/pkg/metadb/datastore/cursor_test.go b/internal/pkg/metadb/cursor_test.go
similarity index 97%
rename from internal/pkg/metadb/datastore/cursor_test.go
rename to internal/pkg/metadb/cursor_test.go
index 327bfbc..8f76b7f 100644
--- a/internal/pkg/metadb/datastore/cursor_test.go
+++ b/internal/pkg/metadb/cursor_test.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package datastore
+package metadb
 
 import (
 	"testing"
diff --git a/internal/pkg/metadb/datastore/datastore_test.go b/internal/pkg/metadb/datastore/datastore_test.go
deleted file mode 100644
index 283dbce..0000000
--- a/internal/pkg/metadb/datastore/datastore_test.go
+++ /dev/null
@@ -1,805 +0,0 @@
-// Copyright 2020 Google LLC
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package datastore
-
-import (
-	"context"
-	"testing"
-	"time"
-
-	"cloud.google.com/go/datastore"
-	"github.com/google/uuid"
-	pb "github.com/googleforgames/open-saves/api"
-	m "github.com/googleforgames/open-saves/internal/pkg/metadb"
-	"github.com/googleforgames/open-saves/internal/pkg/metadb/metadbtest"
-	"github.com/stretchr/testify/assert"
-	"google.golang.org/api/iterator"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
-)
-
-const testTimestampThreshold = 30 * time.Second
-
-func newDriver(ctx context.Context, t *testing.T) *Driver {
-	t.Helper()
-	driver, err := NewDriver(ctx, "triton-for-games-dev")
-	if err != nil {
-		t.Fatalf("Initializing Datastore driver: %v", err)
-	}
-	driver.Namespace = "datastore-unittests"
-	if err := driver.Connect(ctx); err != nil {
-		t.Fatalf("Failed to connect to Datastore: %v", err)
-	}
-	t.Cleanup(func() { driver.Disconnect(ctx) })
-	return driver
-}
-
-func newStoreKey() string {
-	return uuid.New().String() + "_unittest_store"
-}
-
-func newRecordKey() string {
-	return uuid.New().String() + "_unittest_record"
-}
-
-func cloneRecord(r *m.Record) *m.Record {
-	if r == nil {
-		return nil
-	}
-	ret := *r
-	ret.Blob = append([]byte{}, r.Blob...)
-	ret.Properties = make(m.PropertyMap)
-	for k, v := range r.Properties {
-		ret.Properties[k] = v
-	}
-	ret.Tags = append([]string{}, r.Tags...)
-	return &ret
-}
-
-// setupTestStoreRecord creates a new store and inserts a record into it, then registers
-// cleanup functions to delete these test store and record.
-// Passing a nil to record will skip the record insertion.
-func setupTestStoreRecord(ctx context.Context, t *testing.T, driver *Driver, store *m.Store, record *m.Record) (*m.Store, *m.Record) {
-	t.Helper()
-	newStore, err := driver.CreateStore(ctx, store)
-	if err != nil {
-		t.Fatalf("Could not create a new store: %v", err)
-	}
-	t.Cleanup(func() {
-		driver.DeleteStore(ctx, newStore.Key)
-	})
-	metadbtest.AssertEqualStore(t, store, newStore, "CreateStore should return the created store.")
-	var newRecord *m.Record
-	if record != nil {
-		newRecord, err = driver.InsertRecord(ctx, newStore.Key, record)
-		if err != nil {
-			t.Fatalf("Could not create a new record: %v", err)
-		}
-		t.Cleanup(func() {
-			driver.DeleteRecord(ctx, newStore.Key, newRecord.Key)
-		})
-		metadbtest.AssertEqualRecord(t, record, newRecord, "GetRecord should return the exact same record.")
-	}
-	return newStore, newRecord
-}
-
-func setupTestBlobRef(ctx context.Context, t *testing.T, driver *Driver, blob *m.BlobRef) *m.BlobRef {
-	t.Helper()
-	newBlob, err := driver.InsertBlobRef(ctx, blob)
-	if err != nil {
-		t.Fatalf("InsertBlobRef failed: %v", err)
-	}
-	metadbtest.AssertEqualBlobRef(t, blob, newBlob)
-	// save the current namespace as it might change between now and when we delete
-	namespace := driver.Namespace
-	t.Cleanup(func() {
-		// Call the Datastore method directly to avoid Status checking
-		key := datastore.NameKey(blobKind, blob.Key.String(), nil)
-		key.Namespace = namespace
-		driver.client.Delete(ctx, key)
-	})
-	return newBlob
-}
-
-func TestDriver_ConnectDisconnect(t *testing.T) {
-	ctx := context.Background()
-	// Connect() is tested inside newDriver().
-	driver := newDriver(ctx, t)
-	if err := driver.Disconnect(ctx); err != nil {
-		t.Fatalf("Failed to disconnect: %v", err)
-	}
-}
-
-func TestDriver_SimpleCreateGetDeleteStore(t *testing.T) {
-	ctx := context.Background()
-	driver := newDriver(ctx, t)
-	storeKey := newStoreKey()
-	storeName := "SimpleCreateGetDeleteStore" + uuid.New().String()
-	createdAt := time.Date(1988, 4, 16, 8, 6, 5, int(1234*time.Microsecond), time.UTC)
-	store := &m.Store{
-		Key:     storeKey,
-		Name:    storeName,
-		OwnerID: "owner",
-		Tags:    []string{"abc", "def"},
-		Timestamps: m.Timestamps{
-			CreatedAt: createdAt,
-			UpdatedAt: createdAt,
-			Signature: uuid.MustParse("db94be80-e036-4ca8-a9c0-2259b8a67acc"),
-		},
-	}
-	createdStore, err := driver.CreateStore(ctx, store)
-	if err != nil {
-		t.Fatalf("Could not create a new store: %v", err)
-	}
-	metadbtest.AssertEqualStore(t, store, createdStore, "CreateStore should return the created store.")
-
-	store2, err := driver.GetStore(ctx, storeKey)
-	if err != nil {
-		t.Fatalf("Could not get store (%s): %v", storeKey, err)
-	}
-	metadbtest.AssertEqualStore(t, store, store2, "GetStore should return the exact same store.")
-
-	store3, err := driver.FindStoreByName(ctx, storeName)
-	if err != nil {
-		t.Fatalf("Could not fetch store by name (%s): %v", storeName, err)
-	}
-	metadbtest.AssertEqualStore(t, store, store3, "FindStoreByName should return the exact same store.")
-
-	err = driver.DeleteStore(ctx, storeKey)
-	if err != nil {
-		t.Fatalf("Failed to delete a store (%s): %v", storeKey, err)
-	}
-	store4, err := driver.GetStore(ctx, storeKey)
-	if err == nil {
-		t.Fatalf("GetStore didn't return an error after deleting a store: %v", err)
-	}
-	assert.Nil(t, store4, "GetStore should return a nil store.")
-}
-
-func TestDriver_SimpleCreateGetDeleteRecord(t *testing.T) {
-	ctx := context.Background()
-	driver := newDriver(ctx, t)
-	storeKey := newStoreKey()
-	store := &m.Store{
-		Key:     storeKey,
-		Name:    "SimpleCreateGetDeleteRecord",
-		OwnerID: "owner",
-	}
-	setupTestStoreRecord(ctx, t, driver, store, nil)
-
-	recordKey := newRecordKey()
-	blob := []byte{0x54, 0x72, 0x69, 0x74, 0x6f, 0x6e}
-	createdAt := time.Date(1988, 4, 16, 8, 6, 5, int(1234*time.Microsecond), time.UTC)
-	record := &m.Record{
-		Key:      recordKey,
-		Blob:     blob,
-		BlobSize: int64(len(blob)),
-		OwnerID:  "record owner",
-		Tags:     []string{"abc", "def"},
-		Properties: m.PropertyMap{
-			"BoolTP":   {Type: pb.Property_BOOLEAN, BooleanValue: false},
-			"IntTP":    {Type: pb.Property_INTEGER, IntegerValue: 42},
-			"StringTP": {Type: pb.Property_STRING, StringValue: "a string value"},
-		},
-		Timestamps: m.Timestamps{
-			CreatedAt: createdAt,
-			UpdatedAt: createdAt,
-			Signature: uuid.MustParse("89223949-0414-438e-8f5e-3fd9e2d11c1e"),
-		},
-	}
-	expected := cloneRecord(record)
-
-	createdRecord, err := driver.InsertRecord(ctx, storeKey, record)
-	if err != nil {
-		t.Fatalf("Failed to create a new record (%s) in store (%s): %v", recordKey, storeKey, err)
-	}
-	metadbtest.AssertEqualRecord(t, expected, createdRecord, "InsertRecord should return the created record.")
-
-	record2, err := driver.GetRecord(ctx, storeKey, recordKey)
-	if err != nil {
-		t.Fatalf("Failed to get a record (%s) in store (%s): %v", recordKey, storeKey, err)
-	}
-	metadbtest.AssertEqualRecord(t, expected, record2, "GetRecord should return the exact same record.")
-
-	record3, err := driver.InsertRecord(ctx, storeKey, record)
-	assert.NotNil(t, err, "Insert should fail if a record with the same already exists.")
-	assert.Nil(t, record3, "Insert should fail and return nil.")
-
-	err = driver.DeleteRecord(ctx, storeKey, recordKey)
-	assert.Nilf(t, err, "Failed to delete a record (%s) in store (%s): %v", recordKey, storeKey, err)
-
-	record4, err := driver.GetRecord(ctx, storeKey, recordKey)
-	assert.NotNilf(t, err, "GetRecord didn't return an error after deleting a record (%s)", recordKey)
-	assert.Nil(t, record4, "GetRecord should return a nil with error")
-
-	// The test store is deleted inside the cleanup function.
-}
-
-func TestDriver_InsertRecordShouldFailWithNonExistentStore(t *testing.T) {
-	ctx := context.Background()
-	driver := newDriver(ctx, t)
-	storeKey := newStoreKey()
-	record := &m.Record{
-		Key: newRecordKey(),
-	}
-	record1, err := driver.InsertRecord(ctx, storeKey, record)
-	if err == nil {
-		t.Error("InsertRecord should fail if the store doesn't exist.")
-		driver.DeleteRecord(ctx, storeKey, record.Key)
-	} else {
-		assert.Equalf(t, codes.FailedPrecondition, status.Code(err),
-			"InsertRecord should return FailedPrecondition if the store doesn't exist: %v", err)
-	}
-	assert.Nil(t, record1, "InsertRecord should return nil with error")
-}
-
-func TestDriver_DeleteStoreShouldFailWhenNotEmpty(t *testing.T) {
-	ctx := context.Background()
-	driver := newDriver(ctx, t)
-	store := &m.Store{Key: newStoreKey()}
-	record := &m.Record{Key: newRecordKey(), Properties: make(m.PropertyMap)}
-
-	setupTestStoreRecord(ctx, t, driver, store, record)
-
-	if err := driver.DeleteStore(ctx, store.Key); err == nil {
-		t.Error("DeleteStore should fail if the store is not empty.")
-	} else {
-		assert.Equalf(t, codes.FailedPrecondition, status.Code(err),
-			"DeleteStore should return FailedPrecondition if the store is not empty: %v", err)
-	}
-}
-
-func TestDriver_UpdateRecord(t *testing.T) {
-	ctx := context.Background()
-	driver := newDriver(ctx, t)
-
-	storeKey := newStoreKey()
-	store := &m.Store{
-		Key:     storeKey,
-		Name:    "UpdateRecord",
-		OwnerID: "owner",
-	}
-
-	recordKey := newRecordKey()
-	blob := []byte{0x54, 0x72, 0x69, 0x74, 0x6f, 0x6e}
-	record := &m.Record{
-		Key:        recordKey,
-		Blob:       blob,
-		BlobSize:   int64(len(blob)),
-		Properties: make(m.PropertyMap),
-		OwnerID:    "record owner",
-		Tags:       []string{"abc", "def"},
-	}
-	record.Timestamps.NewTimestamps(driver.TimestampPrecision())
-	expected := cloneRecord(record)
-
-	updatedRecord, err := driver.UpdateRecord(ctx, storeKey, recordKey,
-		func(*m.Record) (*m.Record, error) { return nil, nil })
-	assert.NotNil(t, err, "UpdateRecord should return an error if the specified record doesn't exist.")
-	assert.Nil(t, updatedRecord, "UpdateRecord should return nil with error")
-
-	setupTestStoreRecord(ctx, t, driver, store, record)
-
-	record.Tags = append(record.Tags, "ghi")
-	expected.Tags = append(expected.Tags, "ghi")
-	record.OwnerID = "NewOwner"
-	expected.OwnerID = record.OwnerID
-	expected.Timestamps.NewTimestamps(driver.TimestampPrecision())
-
-	// Make sure UpdateRecord doesn't update CreatedAt
-	record.Timestamps.CreatedAt = time.Unix(0, 0)
-	updatedRecord, err = driver.UpdateRecord(ctx, storeKey, recordKey,
-		func(r *m.Record) (*m.Record, error) {
-			r.OwnerID = record.OwnerID
-			r.Tags = record.Tags
-			return r, nil
-		})
-	assert.Nilf(t, err, "Failed to update a record (%s) in store (%s): %v", recordKey, storeKey, err)
-	// Make sure the signatures are different before passing to AssertEqualRecordWithinDuration
-	assert.NotEqual(t, expected.Timestamps.Signature, updatedRecord.Timestamps.Signature)
-	expected.Timestamps.Signature = updatedRecord.Timestamps.Signature
-	metadbtest.AssertEqualRecordWithinDuration(t, expected,
-		updatedRecord, testTimestampThreshold, "Updated record is not the same as original.")
-
-	record2, err := driver.GetRecord(ctx, storeKey, recordKey)
-	if err != nil {
-		t.Fatalf("Failed to get a record (%s) in store (%s): %v", recordKey, storeKey, err)
-	}
-	metadbtest.AssertEqualRecord(t, updatedRecord, record2, "GetRecord should fetch the updated record.")
-}
-
-func TestDriver_DeleteShouldNotFailWithNonExistentKey(t *testing.T) {
-	ctx := context.Background()
-	driver := newDriver(ctx, t)
-	storeKey := newStoreKey()
-	if err := driver.DeleteStore(ctx, storeKey); err != nil {
-		t.Fatalf("DeleteStore failed with a non-existent key: %v", err)
-	}
-	recordKey := newRecordKey()
-	if err := driver.DeleteRecord(ctx, storeKey, recordKey); err != nil {
-		t.Fatalf("DeleteRecord failed with a non-existent key: %v", err)
-	}
-}
-
-func TestDriver_TimestampPrecision(t *testing.T) {
-	ctx := context.Background()
-	driver := newDriver(ctx, t)
-	assert.Equal(t, timestampPrecision, driver.TimestampPrecision())
-}
-
-func TestDriver_SimpleCreateGetDeleteBlobRef(t *testing.T) {
-	ctx := context.Background()
-	driver := newDriver(ctx, t)
-	storeKey := newStoreKey()
-	store := &m.Store{
-		Key:  storeKey,
-		Name: "SimpleCreateGetDeleteBlobRef",
-	}
-	testBlob := []byte{1, 2, 3, 4, 5}
-
-	recordKey := newRecordKey()
-	createdAt := time.Unix(12345, 0)
-	record := &m.Record{
-		Key:        recordKey,
-		Blob:       testBlob,
-		BlobSize:   int64(len(testBlob)),
-		Properties: make(m.PropertyMap),
-		Timestamps: m.Timestamps{
-			CreatedAt: createdAt,
-			UpdatedAt: createdAt,
-			Signature: uuid.New(),
-		},
-	}
-
-	setupTestStoreRecord(ctx, t, driver, store, record)
-	blobKey := uuid.New()
-	origSig := uuid.New()
-	blob := &m.BlobRef{
-		Key:       blobKey,
-		Size:      12345,
-		Status:    m.BlobRefStatusInitializing,
-		StoreKey:  storeKey,
-		RecordKey: recordKey,
-		Timestamps: m.Timestamps{
-			CreatedAt: createdAt,
-			UpdatedAt: createdAt,
-			Signature: origSig,
-		},
-	}
-
-	setupTestBlobRef(ctx, t, driver, blob)
-
-	_, err := driver.GetCurrentBlobRef(ctx, storeKey, recordKey)
-	assert.Equal(t, codes.FailedPrecondition, grpc.Code(err))
-
-	blob2, err := driver.GetBlobRef(ctx, blobKey)
-	if err != nil {
-		t.Errorf("GetBlobRef failed: %v", err)
-	} else {
-		metadbtest.AssertEqualBlobRef(t, blob, blob2)
-	}
-
-	beforePromo := time.Now()
-	promoRecord, promoBlob, err := driver.PromoteBlobRefToCurrent(ctx, blob)
-	if err != nil {
-		t.Errorf("PromoteBlobAsCurrent failed: %v", err)
-	} else {
-		if assert.NotNil(t, promoRecord) {
-			assert.Nil(t, promoRecord.Blob)
-			assert.Equal(t, blobKey, promoRecord.ExternalBlob)
-			assert.Equal(t, blob.Size, promoRecord.BlobSize)
-			assert.True(t, beforePromo.Before(promoRecord.Timestamps.UpdatedAt))
-			assert.NotEqual(t, record.Timestamps.Signature, promoRecord.Timestamps.Signature)
-		}
-		if assert.NotNil(t, promoBlob) {
-			assert.Equal(t, blobKey, promoBlob.Key)
-			assert.Equal(t, m.BlobRefStatusReady, promoBlob.Status)
-			assert.True(t, beforePromo.Before(promoBlob.Timestamps.UpdatedAt))
-			assert.NotEqual(t, origSig, promoBlob.Timestamps.Signature)
-		}
-	}
-
-	currentBlob, err := driver.GetCurrentBlobRef(ctx, storeKey, recordKey)
-	if err != nil {
-		t.Errorf("GetCurrentBlobRef failed: %v", err)
-	} else {
-		metadbtest.AssertEqualBlobRefWithinDuration(t, promoBlob, currentBlob, driver.TimestampPrecision())
-	}
-
-	if err := driver.DeleteBlobRef(ctx, blobKey); err == nil {
-		t.Errorf("DeleteBlobRef should fail on current blob: %v", err)
-	} else {
-		assert.Equal(t, codes.FailedPrecondition, grpc.Code(err))
-	}
-
-	delPendRecord, delPendBlob, err := driver.RemoveBlobFromRecord(ctx, storeKey, recordKey)
-	if err != nil {
-		t.Errorf("RemoveBlobFromRecord failed: %v", err)
-	} else {
-		if assert.NotNil(t, delPendRecord) {
-			assert.Equal(t, uuid.Nil, delPendRecord.ExternalBlob)
-			assert.Empty(t, delPendRecord.Blob)
-			assert.Zero(t, delPendRecord.BlobSize)
-		}
-		if assert.NotNil(t, delPendBlob) {
-			assert.Equal(t, m.BlobRefStatusPendingDeletion, delPendBlob.Status)
-		}
-	}
-
-	assert.NoError(t, driver.DeleteBlobRef(ctx, blobKey))
-
-	deletedBlob, err := driver.GetBlobRef(ctx, blobKey)
-	assert.Nil(t, deletedBlob)
-	assert.Equal(t, codes.NotFound, grpc.Code(err), "GetBlobRef should return NotFound after deletion.")
-}
-
-func TestDriver_SwapBlobRefs(t *testing.T) {
-	ctx := context.Background()
-	driver := newDriver(ctx, t)
-	store := &m.Store{
-		Key:  newStoreKey(),
-		Name: "SwapBlobRefs",
-	}
-
-	record := &m.Record{
-		Key:        newRecordKey(),
-		Properties: make(m.PropertyMap),
-	}
-	setupTestStoreRecord(ctx, t, driver, store, record)
-
-	blob := &m.BlobRef{
-		Key:       uuid.New(),
-		Status:    m.BlobRefStatusInitializing,
-		StoreKey:  store.Key,
-		RecordKey: record.Key,
-	}
-	setupTestBlobRef(ctx, t, driver, blob)
-
-	_, blob, err := driver.PromoteBlobRefToCurrent(ctx, blob)
-	if err != nil {
-		t.Errorf("PromoteBlobRefToCurrent failed: %v", err)
-	}
-
-	newBlob := &m.BlobRef{
-		Key:       uuid.New(),
-		Status:    m.BlobRefStatusInitializing,
-		StoreKey:  store.Key,
-		RecordKey: record.Key,
-	}
-	setupTestBlobRef(ctx, t, driver, newBlob)
-
-	record, newCurrBlob, err := driver.PromoteBlobRefToCurrent(ctx, newBlob)
-	if assert.NoError(t, err) {
-		if assert.NotNil(t, record) {
-			assert.Equal(t, newBlob.Key, record.ExternalBlob)
-		}
-		if assert.NotNil(t, newCurrBlob) {
-			assert.Equal(t, newBlob.Key, newCurrBlob.Key)
-			assert.Equal(t, m.BlobRefStatusReady, newCurrBlob.Status)
-		}
-	}
-
-	oldBlob, err := driver.GetBlobRef(ctx, blob.Key)
-	if assert.NoError(t, err) {
-		if assert.NotNil(t, oldBlob) {
-			assert.Equal(t, blob.Key, oldBlob.Key)
-			assert.Equal(t, m.BlobRefStatusPendingDeletion, oldBlob.Status)
-		}
-	}
-
-	_, _, err = driver.RemoveBlobFromRecord(ctx, store.Key, record.Key)
-	assert.NoError(t, err)
-}
-
-func TestDriver_UpdateBlobRef(t *testing.T) {
-	ctx := context.Background()
-	driver := newDriver(ctx, t)
-	store := &m.Store{
-		Key:  newStoreKey(),
-		Name: "SimpleCreateGetDeleteBlobRef",
-	}
-
-	record := &m.Record{
-		Key:        newRecordKey(),
-		Properties: make(m.PropertyMap),
-	}
-
-	setupTestStoreRecord(ctx, t, driver, store, record)
-	blob := &m.BlobRef{
-		Key:       uuid.New(),
-		Status:    m.BlobRefStatusInitializing,
-		StoreKey:  store.Key,
-		RecordKey: record.Key,
-		Timestamps: m.Timestamps{
-			CreatedAt: time.Unix(123, 0),
-			UpdatedAt: time.Unix(123, 0),
-		},
-	}
-
-	setupTestBlobRef(ctx, t, driver, blob)
-
-	assert.NoError(t, blob.Fail())
-	blob.Timestamps.UpdatedAt = time.Unix(234, 0)
-
-	updatedBlob, err := driver.UpdateBlobRef(ctx, blob)
-	if err != nil {
-		t.Errorf("UpdateBlobRef failed: %v", err)
-	} else {
-		if assert.NotNil(t, updatedBlob) {
-			metadbtest.AssertEqualBlobRef(t, blob, updatedBlob)
-		}
-	}
-
-	receivedBlob, err := driver.GetBlobRef(ctx, blob.Key)
-	if err != nil {
-		t.Errorf("GetBlobRef failed: %v", err)
-	} else {
-		if assert.NotNil(t, receivedBlob) {
-			metadbtest.AssertEqualBlobRef(t, blob, receivedBlob)
-		}
-	}
-}
-
-func TestDriver_BlobInsertShouldFailForNonexistentRecord(t *testing.T) {
-	ctx := context.Background()
-	driver := newDriver(ctx, t)
-	blob := &m.BlobRef{
-		Key:       uuid.New(),
-		StoreKey:  "non-existent" + uuid.New().String(),
-		RecordKey: "non-existent" + uuid.New().String(),
-	}
-
-	insertedBlob, err := driver.InsertBlobRef(ctx, blob)
-	if err == nil {
-		t.Error("InsertBlobRef should fail for a non-existent record.")
-		assert.NoError(t, driver.DeleteBlobRef(ctx, blob.Key))
-	} else {
-		assert.Equal(t, codes.FailedPrecondition, grpc.Code(err))
-		assert.Nil(t, insertedBlob)
-	}
-}
-
-func TestDriver_UpdateRecordWithExternalBlobs(t *testing.T) {
-	ctx := context.Background()
-	driver := newDriver(ctx, t)
-
-	storeKey := newStoreKey()
-	store := &m.Store{
-		Key: storeKey,
-	}
-
-	recordKey := newRecordKey()
-	record := &m.Record{
-		Key:        recordKey,
-		Properties: make(m.PropertyMap),
-	}
-	setupTestStoreRecord(ctx, t, driver, store, record)
-
-	blobKey := uuid.New()
-	blob := &m.BlobRef{
-		Key:       blobKey,
-		Status:    m.BlobRefStatusInitializing,
-		StoreKey:  storeKey,
-		RecordKey: recordKey,
-		Timestamps: m.Timestamps{
-			CreatedAt: time.Unix(123, 0),
-			UpdatedAt: time.Unix(123, 0),
-		},
-	}
-	setupTestBlobRef(ctx, t, driver, blob)
-
-	_, blob, err := driver.PromoteBlobRefToCurrent(ctx, blob)
-	if err != nil {
-		t.Errorf("PromoteBlobRefToCurrent failed: %v", err)
-	}
-
-	_, err = driver.UpdateRecord(ctx, storeKey, recordKey, func(record *m.Record) (*m.Record, error) {
-		record.ExternalBlob = uuid.New()
-		return record, nil
-	})
-	if assert.Error(t, err, "UpdateRecord should fail when ExternalBlob is modified") {
-		assert.Equal(t, codes.Internal, grpc.Code(err))
-	}
-
-	// Make sure ExternalBlob is preserved when not modified
-	newRecord, err := driver.UpdateRecord(ctx, storeKey, recordKey, func(record *m.Record) (*m.Record, error) {
-		// noop
-		return record, nil
-	})
-
-	if assert.NoError(t, err) {
-		if assert.NotNil(t, newRecord) {
-			assert.Equal(t, blobKey, newRecord.ExternalBlob)
-		}
-	}
-
-	actual, err := driver.GetRecord(ctx, storeKey, recordKey)
-	if assert.NoError(t, err) {
-		if assert.NotNil(t, actual) {
-			assert.Equal(t, blobKey, actual.ExternalBlob)
-		}
-	}
-
-	// Check if ExternalBlob is marked for deletion
-	testBlob := []byte{0x54, 0x72, 0x69, 0x74, 0x6f, 0x6e}
-	newRecord, err = driver.UpdateRecord(ctx, storeKey, recordKey, func(record *m.Record) (*m.Record, error) {
-		record.Blob = testBlob
-		record.BlobSize = int64(len(testBlob))
-		return record, nil
-	})
-	if assert.NoError(t, err) {
-		if assert.NotNil(t, newRecord) {
-			assert.Equal(t, testBlob, newRecord.Blob)
-			assert.Equal(t, int64(len(testBlob)), newRecord.BlobSize)
-			assert.Equal(t, uuid.Nil, newRecord.ExternalBlob)
-		}
-	}
-
-	actual, err = driver.GetRecord(ctx, storeKey, recordKey)
-	if assert.NoError(t, err) {
-		if assert.NotNil(t, actual) {
-			assert.Equal(t, testBlob, actual.Blob)
-			assert.Equal(t, int64(len(testBlob)), actual.BlobSize)
-			assert.Equal(t, uuid.Nil, actual.ExternalBlob)
-		}
-	}
-
-	blob, err = driver.GetBlobRef(ctx, blob.Key)
-	if assert.NoError(t, err) {
-		if assert.NotNil(t, blob) {
-			assert.Equal(t, m.BlobRefStatusPendingDeletion, blob.Status)
-		}
-	}
-}
-
-func TestDriver_ListBlobsByStatus(t *testing.T) {
-	ctx := context.Background()
-	driver := newDriver(ctx, t)
-
-	storeKey := newStoreKey()
-	store := &m.Store{
-		Key: storeKey,
-	}
-
-	recordKey := newRecordKey()
-	record := &m.Record{
-		Key:        recordKey,
-		Properties: make(m.PropertyMap),
-	}
-	setupTestStoreRecord(ctx, t, driver, store, record)
-
-	statuses := []m.BlobRefStatus{m.BlobRefStatusError, m.BlobRefStatusInitializing, m.BlobRefStatusPendingDeletion, m.BlobRefStatusPendingDeletion}
-	blobs := []*m.BlobRef{}
-	for i, s := range statuses {
-		blob := &m.BlobRef{
-			Key:       uuid.New(),
-			Status:    s,
-			StoreKey:  storeKey,
-			RecordKey: recordKey,
-			Timestamps: m.Timestamps{
-				UpdatedAt: time.Date(2000, 1, i, 0, 0, 0, 0, time.UTC),
-			},
-		}
-		blobs = append(blobs, blob)
-		setupTestBlobRef(ctx, t, driver, blob)
-	}
-
-	// Should return iterator.Done and nil when not found
-	iter, err := driver.ListBlobRefsByStatus(ctx, m.BlobRefStatusError, time.Date(1999, 1, 1, 0, 0, 0, 0, time.UTC))
-	assert.NoError(t, err)
-	if assert.NotNil(t, iter) {
-		b, err := iter.Next()
-		assert.Equal(t, iterator.Done, err)
-		assert.Nil(t, b)
-	}
-
-	iter, err = driver.ListBlobRefsByStatus(ctx, m.BlobRefStatusPendingDeletion, time.Date(2001, 1, 1, 0, 0, 0, 0, time.UTC))
-	assert.NoError(t, err)
-	if assert.NotNil(t, iter) {
-		// Should return both of the PendingDeletion entries
-		b, err := iter.Next()
-		assert.NoError(t, err)
-		if assert.NotNil(t, b) {
-			metadbtest.AssertEqualBlobRef(t, blobs[2], b)
-		}
-		b, err = iter.Next()
-		assert.NoError(t, err)
-		if assert.NotNil(t, b) {
-			metadbtest.AssertEqualBlobRef(t, blobs[3], b)
-		}
-		b, err = iter.Next()
-		assert.Equal(t, iterator.Done, err)
-		assert.Nil(t, b)
-	}
-}
-
-func TestDriver_DeleteRecordWithExternalBlob(t *testing.T) {
-	ctx := context.Background()
-	driver := newDriver(ctx, t)
-
-	storeKey := newStoreKey()
-	store := &m.Store{
-		Key:  storeKey,
-		Name: t.Name(),
-	}
-
-	recordKey := newRecordKey()
-	record := &m.Record{
-		Key:        recordKey,
-		Tags:       []string{t.Name()},
-		Properties: make(m.PropertyMap),
-	}
-	setupTestStoreRecord(ctx, t, driver, store, record)
-
-	blob := m.NewBlobRef(0, storeKey, recordKey)
-	setupTestBlobRef(ctx, t, driver, blob)
-
-	_, blob, err := driver.PromoteBlobRefToCurrent(ctx, blob)
-	if err != nil {
-		t.Errorf("PromoteBlobRefToCurrent failed: %v", err)
-	}
-
-	if err := driver.DeleteRecord(ctx, storeKey, recordKey); err != nil {
-		t.Errorf("DeleteRecord failed: %v", err)
-	}
-
-	actual, err := driver.GetBlobRef(ctx, blob.Key)
-	assert.NoError(t, err, "GetBlobRef should not return error")
-	if assert.NotNil(t, actual) {
-		assert.Equal(t, m.BlobRefStatusPendingDeletion, actual.Status)
-	}
-}
-
-// This case tests if DeleteRecord deletes a record anyway when
-// the associated BlobRef does not exist and the database is inconsistent.
-func TestDriver_DeleteRecordWithNonExistentBlobRef(t *testing.T) {
-	ctx := context.Background()
-	driver := newDriver(ctx, t)
-
-	storeKey := newStoreKey()
-	store := &m.Store{
-		Key:  storeKey,
-		Name: t.Name(),
-	}
-
-	recordKey := newRecordKey()
-	record := &m.Record{
-		Key:          recordKey,
-		ExternalBlob: uuid.New(),
-		Tags:         []string{t.Name()},
-		Properties:   make(m.PropertyMap),
-	}
-
-	setupTestStoreRecord(ctx, t, driver, store, record)
-
-	actualBlob, err := driver.GetBlobRef(ctx, record.ExternalBlob)
-	assert.Nil(t, actualBlob)
-	if assert.Error(t, err, "GetBlobRef should return error when BlobRef doesn't exist.") {
-		assert.Equal(t, codes.NotFound, grpc.Code(err),
-			"GetBlobRef should return NotFound when BlobRef doesn't exist.")
-	}
-
-	assert.NotEqual(t, uuid.Nil, record.ExternalBlob)
-	assert.NoError(t, driver.DeleteRecord(ctx, storeKey, recordKey),
-		"DeleteRecord should succeed even if ExternalBlob doesn't exist.")
-
-	actualRecord, err := driver.GetRecord(ctx, storeKey, recordKey)
-	assert.Nil(t, actualRecord)
-	if assert.Error(t, err, "GetRecord should return error after DeleteRecord") {
-		assert.Equal(t, codes.NotFound, grpc.Code(err),
-			"GetRecord should return NotFound after DeleteRecord")
-	}
-}
diff --git a/internal/pkg/metadb/metadb_test.go b/internal/pkg/metadb/metadb_test.go
index 22afb57..8df89c3 100644
--- a/internal/pkg/metadb/metadb_test.go
+++ b/internal/pkg/metadb/metadb_test.go
@@ -19,137 +19,817 @@ import (
 	"testing"
 	"time"
 
-	"github.com/golang/mock/gomock"
+	"cloud.google.com/go/datastore"
 	"github.com/google/uuid"
+	pb "github.com/googleforgames/open-saves/api"
 	m "github.com/googleforgames/open-saves/internal/pkg/metadb"
-	mock_metadb "github.com/googleforgames/open-saves/internal/pkg/metadb/mock"
+	"github.com/googleforgames/open-saves/internal/pkg/metadb/metadbtest"
 	"github.com/stretchr/testify/assert"
+	"google.golang.org/api/iterator"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+const (
+	storeKind          = "store"
+	recordKind         = "record"
+	blobKind           = "blob"
+	timestampPrecision = 1 * time.Microsecond
+	timestampTestDelta = 5 * time.Second
+	testProject        = "triton-for-games-dev"
+	testNamespace      = "datastore-unittests"
 )
 
 func TestMetaDB_NewMetaDB(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	mockDriver := mock_metadb.NewMockDriver(ctrl)
-	metadb := m.NewMetaDB(mockDriver)
-	assert.NotNil(t, metadb, "NewMetaDB() should return a non-nil instance.")
+	ctx := context.Background()
+	metaDB, err := m.NewMetaDB(ctx, testProject)
+	assert.NotNil(t, metaDB, "NewMetaDB() should return a non-nil instance.")
+	assert.NoError(t, err, "NewMetaDB should succeed.")
+}
+
+const testTimestampThreshold = 30 * time.Second
+
+func newDatastoreClient(ctx context.Context, t *testing.T) *datastore.Client {
+	client, err := datastore.NewClient(ctx, testProject)
+	if err != nil {
+		t.Fatalf("datastore.NewClient failed: %v", err)
+	}
+	t.Cleanup(func() {
+		client.Close()
+	})
+	return client
 }
 
-// Checks if actual is equal to or after expected, within d.
-func timeEqualOrAfter(expected, actual time.Time, d time.Duration) bool {
-	if expected.Equal(actual) {
-		return true
+func newMetaDB(ctx context.Context, t *testing.T) *m.MetaDB {
+	t.Helper()
+	metaDB, err := m.NewMetaDB(ctx, testProject)
+	if err != nil {
+		t.Fatalf("Initializing MetaDB: %v", err)
 	}
-	return expected.Before(actual) && actual.Sub(expected) < d
+	metaDB.Namespace = testNamespace
+	t.Cleanup(func() { metaDB.Disconnect(ctx) })
+	return metaDB
 }
 
-func TestMetaDB_DriverCalls(t *testing.T) {
+func newStoreKey() string {
+	return uuid.New().String() + "_unittest_store"
+}
+
+func newRecordKey() string {
+	return uuid.New().String() + "_unittest_record"
+}
+
+func cloneRecord(r *m.Record) *m.Record {
+	if r == nil {
+		return nil
+	}
+	ret := *r
+	ret.Blob = append([]byte{}, r.Blob...)
+	ret.Properties = make(m.PropertyMap)
+	for k, v := range r.Properties {
+		ret.Properties[k] = v
+	}
+	ret.Tags = append([]string{}, r.Tags...)
+	return &ret
+}
+
+// setupTestStoreRecord creates a new store and inserts a record into it, then registers
+// cleanup functions to delete these test store and record.
+// Passing a nil to record will skip the record insertion.
+func setupTestStoreRecord(ctx context.Context, t *testing.T, metaDB *m.MetaDB, store *m.Store, record *m.Record) (*m.Store, *m.Record) {
+	t.Helper()
+	newStore, err := metaDB.CreateStore(ctx, store)
+	if err != nil {
+		t.Fatalf("Could not create a new store: %v", err)
+	}
+	t.Cleanup(func() {
+		metaDB.DeleteStore(ctx, newStore.Key)
+	})
+	metadbtest.AssertEqualStore(t, store, newStore, "CreateStore should return the created store.")
+	var newRecord *m.Record
+	if record != nil {
+		newRecord, err = metaDB.InsertRecord(ctx, newStore.Key, record)
+		if err != nil {
+			t.Fatalf("Could not create a new record: %v", err)
+		}
+		t.Cleanup(func() {
+			metaDB.DeleteRecord(ctx, newStore.Key, newRecord.Key)
+		})
+		metadbtest.AssertEqualRecord(t, record, newRecord, "GetRecord should return the exact same record.")
+	}
+	return newStore, newRecord
+}
+
+func setupTestBlobRef(ctx context.Context, t *testing.T, metaDB *m.MetaDB, blob *m.BlobRef) *m.BlobRef {
+	t.Helper()
+	newBlob, err := metaDB.InsertBlobRef(ctx, blob)
+	if err != nil {
+		t.Fatalf("InsertBlobRef failed: %v", err)
+	}
+	metadbtest.AssertEqualBlobRef(t, blob, newBlob)
+	// save the current namespace as it might change between now and when we delete
+	namespace := metaDB.Namespace
+	t.Cleanup(func() {
+		// Call the Datastore method directly to avoid Status checking
+		key := datastore.NameKey(blobKind, blob.Key.String(), nil)
+		key.Namespace = namespace
+		newDatastoreClient(ctx, t).Delete(ctx, key)
+	})
+	return newBlob
+}
+
+func TestMetaDB_Disconnect(t *testing.T) {
 	ctx := context.Background()
-	ctrl := gomock.NewController(t)
-	mockDriver := mock_metadb.NewMockDriver(ctrl)
-	store := new(m.Store)
-	record := new(m.Record)
-	const (
-		key           = "random key string"
-		key2          = "another key string"
-		name          = "random name"
-		timeThreshold = 1 * time.Second
-	)
-	aUUID := uuid.New()
+	metaDB := newMetaDB(ctx, t)
+	if err := metaDB.Disconnect(ctx); err != nil {
+		t.Fatalf("Failed to disconnect: %v", err)
+	}
+}
 
-	metadb := m.NewMetaDB(mockDriver)
-	mockDriver.EXPECT().TimestampPrecision().AnyTimes().Return(time.Microsecond)
+func TestMetaDB_SimpleCreateGetDeleteStore(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+	storeKey := newStoreKey()
+	storeName := "SimpleCreateGetDeleteStore" + uuid.New().String()
+	createdAt := time.Date(1988, 4, 16, 8, 6, 5, int(1234*time.Microsecond), time.UTC)
+	store := &m.Store{
+		Key:     storeKey,
+		Name:    storeName,
+		OwnerID: "owner",
+		Tags:    []string{"abc", "def"},
+		Timestamps: m.Timestamps{
+			CreatedAt: createdAt,
+			UpdatedAt: createdAt,
+			Signature: uuid.MustParse("db94be80-e036-4ca8-a9c0-2259b8a67acc"),
+		},
+	}
+	createdStore, err := metaDB.CreateStore(ctx, store)
+	if err != nil {
+		t.Fatalf("Could not create a new store: %v", err)
+	}
+	metadbtest.AssertEqualStore(t, store, createdStore, "CreateStore should return the created store.")
 
-	mockDriver.EXPECT().Connect(ctx)
-	assert.NoError(t, metadb.Connect(ctx))
+	store2, err := metaDB.GetStore(ctx, storeKey)
+	if err != nil {
+		t.Fatalf("Could not get store (%s): %v", storeKey, err)
+	}
+	metadbtest.AssertEqualStore(t, store, store2, "GetStore should return the exact same store.")
 
-	mockDriver.EXPECT().Disconnect(ctx)
-	assert.NoError(t, metadb.Disconnect(ctx))
+	store3, err := metaDB.FindStoreByName(ctx, storeName)
+	if err != nil {
+		t.Fatalf("Could not fetch store by name (%s): %v", storeName, err)
+	}
+	metadbtest.AssertEqualStore(t, store, store3, "FindStoreByName should return the exact same store.")
 
-	beforeCreate := time.Now()
-	mockDriver.EXPECT().CreateStore(ctx, store).Return(store, nil)
-	createdStore, err := metadb.CreateStore(ctx, store)
-	assert.NoError(t, err)
-	assert.NotNil(t, createdStore)
-	assert.True(t, timeEqualOrAfter(beforeCreate, createdStore.Timestamps.CreatedAt, timeThreshold))
-	assert.True(t, timeEqualOrAfter(beforeCreate, createdStore.Timestamps.UpdatedAt, timeThreshold))
+	err = metaDB.DeleteStore(ctx, storeKey)
+	if err != nil {
+		t.Fatalf("Failed to delete a store (%s): %v", storeKey, err)
+	}
+	store4, err := metaDB.GetStore(ctx, storeKey)
+	if err == nil {
+		t.Fatalf("GetStore didn't return an error after deleting a store: %v", err)
+	}
+	assert.Nil(t, store4, "GetStore should return a nil store.")
+}
 
-	mockDriver.EXPECT().GetStore(ctx, key).Return(store, nil)
-	actualstore, err := metadb.GetStore(ctx, key)
-	assert.Same(t, actualstore, store)
-	assert.NoError(t, err)
+func TestMetaDB_SimpleCreateGetDeleteRecord(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+	storeKey := newStoreKey()
+	store := &m.Store{
+		Key:     storeKey,
+		Name:    "SimpleCreateGetDeleteRecord",
+		OwnerID: "owner",
+	}
+	setupTestStoreRecord(ctx, t, metaDB, store, nil)
 
-	mockDriver.EXPECT().FindStoreByName(ctx, name).Return(store, nil)
-	actualstore, err = metadb.FindStoreByName(ctx, name)
-	assert.Same(t, store, actualstore)
-	assert.NoError(t, err)
+	recordKey := newRecordKey()
+	blob := []byte{0x54, 0x72, 0x69, 0x74, 0x6f, 0x6e}
+	createdAt := time.Date(1988, 4, 16, 8, 6, 5, int(1234*time.Microsecond), time.UTC)
+	record := &m.Record{
+		Key:      recordKey,
+		Blob:     blob,
+		BlobSize: int64(len(blob)),
+		OwnerID:  "record owner",
+		Tags:     []string{"abc", "def"},
+		Properties: m.PropertyMap{
+			"BoolTP":   {Type: pb.Property_BOOLEAN, BooleanValue: false},
+			"IntTP":    {Type: pb.Property_INTEGER, IntegerValue: 42},
+			"StringTP": {Type: pb.Property_STRING, StringValue: "a string value"},
+		},
+		Timestamps: m.Timestamps{
+			CreatedAt: createdAt,
+			UpdatedAt: createdAt,
+			Signature: uuid.MustParse("89223949-0414-438e-8f5e-3fd9e2d11c1e"),
+		},
+	}
+	expected := cloneRecord(record)
 
-	mockDriver.EXPECT().DeleteStore(ctx, key)
-	assert.NoError(t, metadb.DeleteStore(ctx, key))
+	createdRecord, err := metaDB.InsertRecord(ctx, storeKey, record)
+	expected.Timestamps.NewTimestamps(timestampPrecision)
+	// Copy the new signature as we cannot generate the same UUID.
+	expected.Timestamps.Signature = createdRecord.Timestamps.Signature
+	if err != nil {
+		t.Fatalf("Failed to create a new record (%s) in store (%s): %v", recordKey, storeKey, err)
+	}
+	metadbtest.AssertEqualRecordWithinDuration(t, expected, createdRecord,
+		testTimestampThreshold, "InsertRecord should return the created record.")
 
-	beforeInsert := time.Now()
-	mockDriver.EXPECT().InsertRecord(ctx, key, record).Return(record, nil)
-	insertedRecord, err := metadb.InsertRecord(ctx, key, record)
-	assert.NoError(t, err)
-	assert.NotNil(t, insertedRecord)
-	assert.True(t, timeEqualOrAfter(beforeInsert, insertedRecord.Timestamps.CreatedAt, timeThreshold))
-	assert.True(t, timeEqualOrAfter(beforeInsert, insertedRecord.Timestamps.UpdatedAt, timeThreshold))
-
-	// The mock driver cannot verify the function argument
-	// due to a technical limitation: https://github.com/golang/mock/issues/324
-	// We can only pass a nil and test the error case.
-	updatedRecord, err := metadb.UpdateRecord(ctx, key, key2, nil)
-	assert.Equal(t, codes.Internal, grpc.Code(err))
-	assert.Nil(t, updatedRecord)
-
-	mockDriver.EXPECT().GetRecord(ctx, key, key2).Return(record, nil)
-	actualrecord, err := metadb.GetRecord(ctx, key, key2)
-	assert.Same(t, actualrecord, record)
-	assert.NoError(t, err)
+	// Use the updated timestamps for the subsequent checks
+	expected.Timestamps = createdRecord.Timestamps
+	record2, err := metaDB.GetRecord(ctx, storeKey, recordKey)
+	if err != nil {
+		t.Fatalf("Failed to get a record (%s) in store (%s): %v", recordKey, storeKey, err)
+	}
+	metadbtest.AssertEqualRecord(t, expected, record2, "GetRecord should return the exact same record.")
 
-	mockDriver.EXPECT().DeleteRecord(ctx, key, key2)
-	assert.NoError(t, metadb.DeleteRecord(ctx, key, key2))
+	record3, err := metaDB.InsertRecord(ctx, storeKey, record)
+	assert.NotNil(t, err, "Insert should fail if a record with the same already exists.")
+	assert.Nil(t, record3, "Insert should fail and return nil.")
 
-	blob := new(m.BlobRef)
+	err = metaDB.DeleteRecord(ctx, storeKey, recordKey)
+	assert.Nilf(t, err, "Failed to delete a record (%s) in store (%s): %v", recordKey, storeKey, err)
 
-	mockDriver.EXPECT().InsertBlobRef(ctx, blob).Return(blob, nil)
-	actualBlob, err := metadb.InsertBlobRef(ctx, blob)
-	assert.Same(t, blob, actualBlob)
-	assert.NoError(t, err)
+	record4, err := metaDB.GetRecord(ctx, storeKey, recordKey)
+	assert.NotNilf(t, err, "GetRecord didn't return an error after deleting a record (%s)", recordKey)
+	assert.Nil(t, record4, "GetRecord should return a nil with error")
 
-	mockDriver.EXPECT().UpdateBlobRef(ctx, blob).Return(blob, nil)
-	actualBlob, err = metadb.UpdateBlobRef(ctx, blob)
-	assert.Same(t, blob, actualBlob)
-	assert.NoError(t, err)
+	// The test store is deleted inside the cleanup function.
+}
 
-	mockDriver.EXPECT().GetBlobRef(ctx, aUUID).Return(blob, nil)
-	actualBlob, err = metadb.GetBlobRef(ctx, aUUID)
-	assert.Same(t, blob, actualBlob)
-	assert.NoError(t, err)
+func TestMetaDB_InsertRecordShouldFailWithNonExistentStore(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+	storeKey := newStoreKey()
+	record := &m.Record{
+		Key: newRecordKey(),
+	}
+	record1, err := metaDB.InsertRecord(ctx, storeKey, record)
+	if err == nil {
+		t.Error("InsertRecord should fail if the store doesn't exist.")
+		metaDB.DeleteRecord(ctx, storeKey, record.Key)
+	} else {
+		assert.Equalf(t, codes.FailedPrecondition, status.Code(err),
+			"InsertRecord should return FailedPrecondition if the store doesn't exist: %v", err)
+	}
+	assert.Nil(t, record1, "InsertRecord should return nil with error")
+}
+
+func TestMetaDB_DeleteStoreShouldFailWhenNotEmpty(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+	store := &m.Store{Key: newStoreKey()}
+	record := &m.Record{Key: newRecordKey(), Properties: make(m.PropertyMap)}
+
+	setupTestStoreRecord(ctx, t, metaDB, store, record)
+
+	if err := metaDB.DeleteStore(ctx, store.Key); err == nil {
+		t.Error("DeleteStore should fail if the store is not empty.")
+	} else {
+		assert.Equalf(t, codes.FailedPrecondition, status.Code(err),
+			"DeleteStore should return FailedPrecondition if the store is not empty: %v", err)
+	}
+}
+
+func TestMetaDB_UpdateRecord(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+
+	storeKey := newStoreKey()
+	store := &m.Store{
+		Key:     storeKey,
+		Name:    "UpdateRecord",
+		OwnerID: "owner",
+	}
+
+	recordKey := newRecordKey()
+	blob := []byte{0x54, 0x72, 0x69, 0x74, 0x6f, 0x6e}
+	record := &m.Record{
+		Key:        recordKey,
+		Blob:       blob,
+		BlobSize:   int64(len(blob)),
+		Properties: make(m.PropertyMap),
+		OwnerID:    "record owner",
+		Tags:       []string{"abc", "def"},
+	}
+	record.Timestamps.NewTimestamps(timestampPrecision)
+	expected := cloneRecord(record)
+
+	updatedRecord, err := metaDB.UpdateRecord(ctx, storeKey, recordKey,
+		func(*m.Record) (*m.Record, error) { return nil, nil })
+	assert.NotNil(t, err, "UpdateRecord should return an error if the specified record doesn't exist.")
+	assert.Nil(t, updatedRecord, "UpdateRecord should return nil with error")
+
+	setupTestStoreRecord(ctx, t, metaDB, store, record)
+
+	record.Tags = append(record.Tags, "ghi")
+	expected.Tags = append(expected.Tags, "ghi")
+	record.OwnerID = "NewOwner"
+	expected.OwnerID = record.OwnerID
+	expected.Timestamps.NewTimestamps(timestampPrecision)
+
+	// Make sure UpdateRecord doesn't update CreatedAt
+	record.Timestamps.CreatedAt = time.Unix(0, 0)
+	updatedRecord, err = metaDB.UpdateRecord(ctx, storeKey, recordKey,
+		func(r *m.Record) (*m.Record, error) {
+			r.OwnerID = record.OwnerID
+			r.Tags = record.Tags
+			return r, nil
+		})
+	assert.Nilf(t, err, "Failed to update a record (%s) in store (%s): %v", recordKey, storeKey, err)
+	// Make sure the signatures are different before passing to AssertEqualRecordWithinDuration
+	assert.NotEqual(t, expected.Timestamps.Signature, updatedRecord.Timestamps.Signature)
+	expected.Timestamps.Signature = updatedRecord.Timestamps.Signature
+	metadbtest.AssertEqualRecordWithinDuration(t, expected,
+		updatedRecord, testTimestampThreshold, "Updated record is not the same as original.")
+
+	record2, err := metaDB.GetRecord(ctx, storeKey, recordKey)
+	if err != nil {
+		t.Fatalf("Failed to get a record (%s) in store (%s): %v", recordKey, storeKey, err)
+	}
+	metadbtest.AssertEqualRecord(t, updatedRecord, record2, "GetRecord should fetch the updated record.")
+}
+
+func TestMetaDB_DeleteShouldNotFailWithNonExistentKey(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+	storeKey := newStoreKey()
+	if err := metaDB.DeleteStore(ctx, storeKey); err != nil {
+		t.Fatalf("DeleteStore failed with a non-existent key: %v", err)
+	}
+	recordKey := newRecordKey()
+	if err := metaDB.DeleteRecord(ctx, storeKey, recordKey); err != nil {
+		t.Fatalf("DeleteRecord failed with a non-existent key: %v", err)
+	}
+}
+
+func TestMetaDB_SimpleCreateGetDeleteBlobRef(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+	storeKey := newStoreKey()
+	store := &m.Store{
+		Key:  storeKey,
+		Name: "SimpleCreateGetDeleteBlobRef",
+	}
+	testBlob := []byte{1, 2, 3, 4, 5}
+
+	recordKey := newRecordKey()
+	createdAt := time.Unix(12345, 0)
+	record := &m.Record{
+		Key:        recordKey,
+		Blob:       testBlob,
+		BlobSize:   int64(len(testBlob)),
+		Properties: make(m.PropertyMap),
+		Timestamps: m.Timestamps{
+			CreatedAt: createdAt,
+			UpdatedAt: createdAt,
+			Signature: uuid.New(),
+		},
+	}
+
+	setupTestStoreRecord(ctx, t, metaDB, store, record)
+	blobKey := uuid.New()
+	origSig := uuid.New()
+	blob := &m.BlobRef{
+		Key:       blobKey,
+		Size:      12345,
+		Status:    m.BlobRefStatusInitializing,
+		StoreKey:  storeKey,
+		RecordKey: recordKey,
+		Timestamps: m.Timestamps{
+			CreatedAt: createdAt,
+			UpdatedAt: createdAt,
+			Signature: origSig,
+		},
+	}
+
+	setupTestBlobRef(ctx, t, metaDB, blob)
+
+	_, err := metaDB.GetCurrentBlobRef(ctx, storeKey, recordKey)
+	assert.Equal(t, codes.FailedPrecondition, grpc.Code(err))
+
+	blob2, err := metaDB.GetBlobRef(ctx, blobKey)
+	if err != nil {
+		t.Errorf("GetBlobRef failed: %v", err)
+	} else {
+		metadbtest.AssertEqualBlobRef(t, blob, blob2)
+	}
 
-	mockDriver.EXPECT().GetCurrentBlobRef(ctx, key, key2).Return(blob, nil)
-	actualBlob, err = metadb.GetCurrentBlobRef(ctx, key, key2)
-	assert.Same(t, blob, actualBlob)
+	beforePromo := time.Now()
+	promoRecord, promoBlob, err := metaDB.PromoteBlobRefToCurrent(ctx, blob)
+	if err != nil {
+		t.Errorf("PromoteBlobAsCurrent failed: %v", err)
+	} else {
+		if assert.NotNil(t, promoRecord) {
+			assert.Nil(t, promoRecord.Blob)
+			assert.Equal(t, blobKey, promoRecord.ExternalBlob)
+			assert.Equal(t, blob.Size, promoRecord.BlobSize)
+			assert.True(t, beforePromo.Before(promoRecord.Timestamps.UpdatedAt))
+			assert.NotEqual(t, record.Timestamps.Signature, promoRecord.Timestamps.Signature)
+		}
+		if assert.NotNil(t, promoBlob) {
+			assert.Equal(t, blobKey, promoBlob.Key)
+			assert.Equal(t, m.BlobRefStatusReady, promoBlob.Status)
+			assert.True(t, beforePromo.Before(promoBlob.Timestamps.UpdatedAt))
+			assert.NotEqual(t, origSig, promoBlob.Timestamps.Signature)
+		}
+	}
+
+	currentBlob, err := metaDB.GetCurrentBlobRef(ctx, storeKey, recordKey)
+	if err != nil {
+		t.Errorf("GetCurrentBlobRef failed: %v", err)
+	} else {
+		metadbtest.AssertEqualBlobRefWithinDuration(t, promoBlob, currentBlob, timestampPrecision)
+	}
+
+	if err := metaDB.DeleteBlobRef(ctx, blobKey); err == nil {
+		t.Errorf("DeleteBlobRef should fail on current blob: %v", err)
+	} else {
+		assert.Equal(t, codes.FailedPrecondition, grpc.Code(err))
+	}
+
+	delPendRecord, delPendBlob, err := metaDB.RemoveBlobFromRecord(ctx, storeKey, recordKey)
+	if err != nil {
+		t.Errorf("RemoveBlobFromRecord failed: %v", err)
+	} else {
+		if assert.NotNil(t, delPendRecord) {
+			assert.Equal(t, uuid.Nil, delPendRecord.ExternalBlob)
+			assert.Empty(t, delPendRecord.Blob)
+			assert.Zero(t, delPendRecord.BlobSize)
+		}
+		if assert.NotNil(t, delPendBlob) {
+			assert.Equal(t, m.BlobRefStatusPendingDeletion, delPendBlob.Status)
+		}
+	}
+
+	assert.NoError(t, metaDB.DeleteBlobRef(ctx, blobKey))
+
+	deletedBlob, err := metaDB.GetBlobRef(ctx, blobKey)
+	assert.Nil(t, deletedBlob)
+	assert.Equal(t, codes.NotFound, grpc.Code(err), "GetBlobRef should return NotFound after deletion.")
+}
+
+func TestMetaDB_SwapBlobRefs(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+	store := &m.Store{
+		Key:  newStoreKey(),
+		Name: "SwapBlobRefs",
+	}
+
+	record := &m.Record{
+		Key:        newRecordKey(),
+		Properties: make(m.PropertyMap),
+	}
+	setupTestStoreRecord(ctx, t, metaDB, store, record)
+
+	blob := &m.BlobRef{
+		Key:       uuid.New(),
+		Status:    m.BlobRefStatusInitializing,
+		StoreKey:  store.Key,
+		RecordKey: record.Key,
+	}
+	setupTestBlobRef(ctx, t, metaDB, blob)
+
+	_, blob, err := metaDB.PromoteBlobRefToCurrent(ctx, blob)
+	if err != nil {
+		t.Errorf("PromoteBlobRefToCurrent failed: %v", err)
+	}
+
+	newBlob := &m.BlobRef{
+		Key:       uuid.New(),
+		Status:    m.BlobRefStatusInitializing,
+		StoreKey:  store.Key,
+		RecordKey: record.Key,
+	}
+	setupTestBlobRef(ctx, t, metaDB, newBlob)
+
+	record, newCurrBlob, err := metaDB.PromoteBlobRefToCurrent(ctx, newBlob)
+	if assert.NoError(t, err) {
+		if assert.NotNil(t, record) {
+			assert.Equal(t, newBlob.Key, record.ExternalBlob)
+		}
+		if assert.NotNil(t, newCurrBlob) {
+			assert.Equal(t, newBlob.Key, newCurrBlob.Key)
+			assert.Equal(t, m.BlobRefStatusReady, newCurrBlob.Status)
+		}
+	}
+
+	oldBlob, err := metaDB.GetBlobRef(ctx, blob.Key)
+	if assert.NoError(t, err) {
+		if assert.NotNil(t, oldBlob) {
+			assert.Equal(t, blob.Key, oldBlob.Key)
+			assert.Equal(t, m.BlobRefStatusPendingDeletion, oldBlob.Status)
+		}
+	}
+
+	_, _, err = metaDB.RemoveBlobFromRecord(ctx, store.Key, record.Key)
 	assert.NoError(t, err)
+}
+
+func TestMetaDB_UpdateBlobRef(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+	store := &m.Store{
+		Key:  newStoreKey(),
+		Name: "SimpleCreateGetDeleteBlobRef",
+	}
+
+	record := &m.Record{
+		Key:        newRecordKey(),
+		Properties: make(m.PropertyMap),
+	}
+
+	setupTestStoreRecord(ctx, t, metaDB, store, record)
+	blob := &m.BlobRef{
+		Key:       uuid.New(),
+		Status:    m.BlobRefStatusInitializing,
+		StoreKey:  store.Key,
+		RecordKey: record.Key,
+		Timestamps: m.Timestamps{
+			CreatedAt: time.Unix(123, 0),
+			UpdatedAt: time.Unix(123, 0),
+		},
+	}
+
+	setupTestBlobRef(ctx, t, metaDB, blob)
+
+	assert.NoError(t, blob.Fail())
+	blob.Timestamps.UpdatedAt = time.Unix(234, 0)
+
+	updatedBlob, err := metaDB.UpdateBlobRef(ctx, blob)
+	if err != nil {
+		t.Errorf("UpdateBlobRef failed: %v", err)
+	} else {
+		if assert.NotNil(t, updatedBlob) {
+			metadbtest.AssertEqualBlobRef(t, blob, updatedBlob)
+		}
+	}
+
+	receivedBlob, err := metaDB.GetBlobRef(ctx, blob.Key)
+	if err != nil {
+		t.Errorf("GetBlobRef failed: %v", err)
+	} else {
+		if assert.NotNil(t, receivedBlob) {
+			metadbtest.AssertEqualBlobRef(t, blob, receivedBlob)
+		}
+	}
+}
+
+func TestMetaDB_BlobInsertShouldFailForNonexistentRecord(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+	blob := &m.BlobRef{
+		Key:       uuid.New(),
+		StoreKey:  "non-existent" + uuid.New().String(),
+		RecordKey: "non-existent" + uuid.New().String(),
+	}
+
+	insertedBlob, err := metaDB.InsertBlobRef(ctx, blob)
+	if err == nil {
+		t.Error("InsertBlobRef should fail for a non-existent record.")
+		assert.NoError(t, metaDB.DeleteBlobRef(ctx, blob.Key))
+	} else {
+		assert.Equal(t, codes.FailedPrecondition, grpc.Code(err))
+		assert.Nil(t, insertedBlob)
+	}
+}
+
+func TestMetaDB_UpdateRecordWithExternalBlobs(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+
+	storeKey := newStoreKey()
+	store := &m.Store{
+		Key: storeKey,
+	}
+
+	recordKey := newRecordKey()
+	record := &m.Record{
+		Key:        recordKey,
+		Properties: make(m.PropertyMap),
+	}
+	setupTestStoreRecord(ctx, t, metaDB, store, record)
+
+	blobKey := uuid.New()
+	blob := &m.BlobRef{
+		Key:       blobKey,
+		Status:    m.BlobRefStatusInitializing,
+		StoreKey:  storeKey,
+		RecordKey: recordKey,
+		Timestamps: m.Timestamps{
+			CreatedAt: time.Unix(123, 0),
+			UpdatedAt: time.Unix(123, 0),
+		},
+	}
+	setupTestBlobRef(ctx, t, metaDB, blob)
+
+	_, blob, err := metaDB.PromoteBlobRefToCurrent(ctx, blob)
+	if err != nil {
+		t.Errorf("PromoteBlobRefToCurrent failed: %v", err)
+	}
+
+	_, err = metaDB.UpdateRecord(ctx, storeKey, recordKey, func(record *m.Record) (*m.Record, error) {
+		record.ExternalBlob = uuid.New()
+		return record, nil
+	})
+	if assert.Error(t, err, "UpdateRecord should fail when ExternalBlob is modified") {
+		assert.Equal(t, codes.Internal, grpc.Code(err))
+	}
+
+	// Make sure ExternalBlob is preserved when not modified
+	newRecord, err := metaDB.UpdateRecord(ctx, storeKey, recordKey, func(record *m.Record) (*m.Record, error) {
+		// noop
+		return record, nil
+	})
 
-	mockDriver.EXPECT().PromoteBlobRefToCurrent(ctx, blob).Return(record, blob, nil)
-	actualrecord, actualBlob, err = metadb.PromoteBlobRefToCurrent(ctx, blob)
-	assert.Same(t, record, actualrecord)
-	assert.Same(t, blob, actualBlob)
+	if assert.NoError(t, err) {
+		if assert.NotNil(t, newRecord) {
+			assert.Equal(t, blobKey, newRecord.ExternalBlob)
+		}
+	}
+
+	actual, err := metaDB.GetRecord(ctx, storeKey, recordKey)
+	if assert.NoError(t, err) {
+		if assert.NotNil(t, actual) {
+			assert.Equal(t, blobKey, actual.ExternalBlob)
+		}
+	}
+
+	// Check if ExternalBlob is marked for deletion
+	testBlob := []byte{0x54, 0x72, 0x69, 0x74, 0x6f, 0x6e}
+	newRecord, err = metaDB.UpdateRecord(ctx, storeKey, recordKey, func(record *m.Record) (*m.Record, error) {
+		record.Blob = testBlob
+		record.BlobSize = int64(len(testBlob))
+		return record, nil
+	})
+	if assert.NoError(t, err) {
+		if assert.NotNil(t, newRecord) {
+			assert.Equal(t, testBlob, newRecord.Blob)
+			assert.Equal(t, int64(len(testBlob)), newRecord.BlobSize)
+			assert.Equal(t, uuid.Nil, newRecord.ExternalBlob)
+		}
+	}
+
+	actual, err = metaDB.GetRecord(ctx, storeKey, recordKey)
+	if assert.NoError(t, err) {
+		if assert.NotNil(t, actual) {
+			assert.Equal(t, testBlob, actual.Blob)
+			assert.Equal(t, int64(len(testBlob)), actual.BlobSize)
+			assert.Equal(t, uuid.Nil, actual.ExternalBlob)
+		}
+	}
+
+	blob, err = metaDB.GetBlobRef(ctx, blob.Key)
+	if assert.NoError(t, err) {
+		if assert.NotNil(t, blob) {
+			assert.Equal(t, m.BlobRefStatusPendingDeletion, blob.Status)
+		}
+	}
+}
+
+func TestMetaDB_ListBlobsByStatus(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+	client := newDatastoreClient(ctx, t)
+
+	storeKey := newStoreKey()
+	store := &m.Store{
+		Key: storeKey,
+	}
+
+	recordKey := newRecordKey()
+	record := &m.Record{
+		Key:        recordKey,
+		Properties: make(m.PropertyMap),
+	}
+	setupTestStoreRecord(ctx, t, metaDB, store, record)
+
+	statuses := []m.BlobRefStatus{m.BlobRefStatusError, m.BlobRefStatusInitializing, m.BlobRefStatusPendingDeletion, m.BlobRefStatusPendingDeletion}
+	blobs := []*m.BlobRef{}
+	for i, s := range statuses {
+		blob := &m.BlobRef{
+			Key:       uuid.New(),
+			Status:    s,
+			StoreKey:  storeKey,
+			RecordKey: recordKey,
+		}
+		blobs = append(blobs, blob)
+		setupTestBlobRef(ctx, t, metaDB, blob)
+
+		// Update the timestamps here because MetaDB automatically sets UpdatesAt
+		blob.Timestamps.UpdatedAt = time.Date(2000, 1, i, 0, 0, 0, 0, time.UTC)
+		bKey := datastore.NameKey(blobKind, blob.Key.String(), nil)
+		bKey.Namespace = metaDB.Namespace
+		if _, err := client.Put(ctx, bKey, blob); err != nil {
+			t.Fatalf("Failed to change UpdatedAt for %v: %v", blob.Key, err)
+		}
+	}
+
+	// Should return iterator.Done and nil when not found
+	iter, err := metaDB.ListBlobRefsByStatus(ctx, m.BlobRefStatusError, time.Date(1999, 1, 1, 0, 0, 0, 0, time.UTC))
 	assert.NoError(t, err)
+	if assert.NotNil(t, iter) {
+		b, err := iter.Next()
+		assert.Equal(t, iterator.Done, err)
+		assert.Nil(t, b)
+	}
 
-	mockDriver.EXPECT().RemoveBlobFromRecord(ctx, key, key2).Return(record, blob, nil)
-	actualrecord, actualBlob, err = metadb.RemoveBlobFromRecord(ctx, key, key2)
-	assert.Same(t, record, actualrecord)
-	assert.Same(t, blob, actualBlob)
+	iter, err = metaDB.ListBlobRefsByStatus(ctx, m.BlobRefStatusPendingDeletion, time.Date(2001, 1, 1, 0, 0, 0, 0, time.UTC))
 	assert.NoError(t, err)
+	if assert.NotNil(t, iter) {
+		// Should return both of the PendingDeletion entries
+		b, err := iter.Next()
+		assert.NoError(t, err)
+		if assert.NotNil(t, b) {
+			metadbtest.AssertEqualBlobRef(t, blobs[2], b)
+		}
+		b, err = iter.Next()
+		assert.NoError(t, err)
+		if assert.NotNil(t, b) {
+			metadbtest.AssertEqualBlobRef(t, blobs[3], b)
+		}
+		b, err = iter.Next()
+		assert.Equal(t, iterator.Done, err)
+		assert.Nil(t, b)
+	}
+}
+
+func TestMetaDB_DeleteRecordWithExternalBlob(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
 
-	mockDriver.EXPECT().DeleteBlobRef(ctx, aUUID)
-	assert.NoError(t, metadb.DeleteBlobRef(ctx, aUUID))
+	storeKey := newStoreKey()
+	store := &m.Store{
+		Key:  storeKey,
+		Name: t.Name(),
+	}
 
-	olderThan := time.Unix(123456, 0)
-	mockDriver.EXPECT().ListBlobRefsByStatus(ctx, m.BlobRefStatusPendingDeletion, olderThan).Return(nil, nil)
-	iter, err := metadb.ListBlobRefsByStatus(ctx, m.BlobRefStatusPendingDeletion, olderThan)
-	assert.Nil(t, iter)
-	assert.NoError(t, err)
+	recordKey := newRecordKey()
+	record := &m.Record{
+		Key:        recordKey,
+		Tags:       []string{t.Name()},
+		Properties: make(m.PropertyMap),
+	}
+	setupTestStoreRecord(ctx, t, metaDB, store, record)
+
+	blob := m.NewBlobRef(0, storeKey, recordKey)
+	setupTestBlobRef(ctx, t, metaDB, blob)
+
+	_, blob, err := metaDB.PromoteBlobRefToCurrent(ctx, blob)
+	if err != nil {
+		t.Errorf("PromoteBlobRefToCurrent failed: %v", err)
+	}
+
+	if err := metaDB.DeleteRecord(ctx, storeKey, recordKey); err != nil {
+		t.Errorf("DeleteRecord failed: %v", err)
+	}
+
+	actual, err := metaDB.GetBlobRef(ctx, blob.Key)
+	assert.NoError(t, err, "GetBlobRef should not return error")
+	if assert.NotNil(t, actual) {
+		assert.Equal(t, m.BlobRefStatusPendingDeletion, actual.Status)
+	}
+}
+
+// This case tests if DeleteRecord deletes a record anyway when
+// the associated BlobRef does not exist and the database is inconsistent.
+func TestMetaDB_DeleteRecordWithNonExistentBlobRef(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+
+	storeKey := newStoreKey()
+	store := &m.Store{
+		Key:  storeKey,
+		Name: t.Name(),
+	}
+
+	recordKey := newRecordKey()
+	record := &m.Record{
+		Key:          recordKey,
+		ExternalBlob: uuid.New(),
+		Tags:         []string{t.Name()},
+		Properties:   make(m.PropertyMap),
+	}
+
+	setupTestStoreRecord(ctx, t, metaDB, store, record)
+
+	actualBlob, err := metaDB.GetBlobRef(ctx, record.ExternalBlob)
+	assert.Nil(t, actualBlob)
+	if assert.Error(t, err, "GetBlobRef should return error when BlobRef doesn't exist.") {
+		assert.Equal(t, codes.NotFound, grpc.Code(err),
+			"GetBlobRef should return NotFound when BlobRef doesn't exist.")
+	}
+
+	assert.NotEqual(t, uuid.Nil, record.ExternalBlob)
+	assert.NoError(t, metaDB.DeleteRecord(ctx, storeKey, recordKey),
+		"DeleteRecord should succeed even if ExternalBlob doesn't exist.")
+
+	actualRecord, err := metaDB.GetRecord(ctx, storeKey, recordKey)
+	assert.Nil(t, actualRecord)
+	if assert.Error(t, err, "GetRecord should return error after DeleteRecord") {
+		assert.Equal(t, codes.NotFound, grpc.Code(err),
+			"GetRecord should return NotFound after DeleteRecord")
+	}
 }