diff --git a/v23/services/syncbase/nosql/service.vdl b/v23/services/syncbase/nosql/service.vdl
index 0f56ba3..d23c9c4 100644
--- a/v23/services/syncbase/nosql/service.vdl
+++ b/v23/services/syncbase/nosql/service.vdl
@@ -264,7 +264,7 @@
 	// fetched before the lower priority ones. However an ongoing blob
 	// transfer is not interrupted. Status updates are streamed back to the
 	// client as fetch is in progress.
-	FetchBlob(br BlobRef, priority uint64) stream<_, FetchStatus> error {access.Read}
+	FetchBlob(br BlobRef, priority uint64) stream<_, BlobFetchStatus> error {access.Read}
 
 	// PinBlob locally pins the blob so that it is not evicted.
 	PinBlob(br BlobRef) error {access.Write}
@@ -322,4 +322,5 @@
 	ReadOnlyBatch() {"en": "batch is read-only"}
 	ConcurrentBatch() {"en": "concurrent batch"}
 	SchemaVersionMismatch() {"en": "actual schema version does not match the provided one"}
+	BlobNotCommitted() {"en": "blob is not yet committed"}
 )
diff --git a/v23/services/syncbase/nosql/service.vdl.go b/v23/services/syncbase/nosql/service.vdl.go
index 624036c..7318ff2 100644
--- a/v23/services/syncbase/nosql/service.vdl.go
+++ b/v23/services/syncbase/nosql/service.vdl.go
@@ -30,6 +30,7 @@
 	ErrReadOnlyBatch         = verror.Register("v.io/syncbase/v23/services/syncbase/nosql.ReadOnlyBatch", verror.NoRetry, "{1:}{2:} batch is read-only")
 	ErrConcurrentBatch       = verror.Register("v.io/syncbase/v23/services/syncbase/nosql.ConcurrentBatch", verror.NoRetry, "{1:}{2:} concurrent batch")
 	ErrSchemaVersionMismatch = verror.Register("v.io/syncbase/v23/services/syncbase/nosql.SchemaVersionMismatch", verror.NoRetry, "{1:}{2:} actual schema version does not match the provided one")
+	ErrBlobNotCommitted      = verror.Register("v.io/syncbase/v23/services/syncbase/nosql.BlobNotCommitted", verror.NoRetry, "{1:}{2:} blob is not yet committed")
 )
 
 func init() {
@@ -38,6 +39,7 @@
 	i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrReadOnlyBatch.ID), "{1:}{2:} batch is read-only")
 	i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrConcurrentBatch.ID), "{1:}{2:} concurrent batch")
 	i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrSchemaVersionMismatch.ID), "{1:}{2:} actual schema version does not match the provided one")
+	i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(ErrBlobNotCommitted.ID), "{1:}{2:} blob is not yet committed")
 }
 
 // NewErrBoundToBatch returns an error with the ErrBoundToBatch ID.
@@ -65,6 +67,11 @@
 	return verror.New(ErrSchemaVersionMismatch, ctx)
 }
 
+// NewErrBlobNotCommitted returns an error with the ErrBlobNotCommitted ID.
+func NewErrBlobNotCommitted(ctx *context.T) error {
+	return verror.New(ErrBlobNotCommitted, ctx)
+}
+
 // DatabaseWatcherClientMethods is the client interface
 // containing DatabaseWatcher methods.
 //
@@ -840,7 +847,7 @@
 		Advance() bool
 		// Value returns the item that was staged by Advance.  May panic if Advance
 		// returned false or was not called.  Never blocks.
-		Value() FetchStatus
+		Value() BlobFetchStatus
 		// Err returns any error encountered by Advance.  Never blocks.
 		Err() error
 	}
@@ -864,13 +871,13 @@
 
 type implBlobManagerFetchBlobClientCall struct {
 	rpc.ClientCall
-	valRecv FetchStatus
+	valRecv BlobFetchStatus
 	errRecv error
 }
 
 func (c *implBlobManagerFetchBlobClientCall) RecvStream() interface {
 	Advance() bool
-	Value() FetchStatus
+	Value() BlobFetchStatus
 	Err() error
 } {
 	return implBlobManagerFetchBlobClientCallRecv{c}
@@ -881,11 +888,11 @@
 }
 
 func (c implBlobManagerFetchBlobClientCallRecv) Advance() bool {
-	c.c.valRecv = FetchStatus{}
+	c.c.valRecv = BlobFetchStatus{}
 	c.c.errRecv = c.c.Recv(&c.c.valRecv)
 	return c.c.errRecv == nil
 }
-func (c implBlobManagerFetchBlobClientCallRecv) Value() FetchStatus {
+func (c implBlobManagerFetchBlobClientCallRecv) Value() BlobFetchStatus {
 	return c.c.valRecv
 }
 func (c implBlobManagerFetchBlobClientCallRecv) Err() error {
@@ -1260,7 +1267,7 @@
 		// Send places the item onto the output stream.  Returns errors encountered
 		// while sending.  Blocks if there is no buffer space; will unblock when
 		// buffer space is available.
-		Send(item FetchStatus) error
+		Send(item BlobFetchStatus) error
 	}
 }
 
@@ -1283,7 +1290,7 @@
 
 // SendStream returns the send side of the BlobManager.FetchBlob server stream.
 func (s *BlobManagerFetchBlobServerCallStub) SendStream() interface {
-	Send(item FetchStatus) error
+	Send(item BlobFetchStatus) error
 } {
 	return implBlobManagerFetchBlobServerCallSend{s}
 }
@@ -1292,7 +1299,7 @@
 	s *BlobManagerFetchBlobServerCallStub
 }
 
-func (s implBlobManagerFetchBlobServerCallSend) Send(item FetchStatus) error {
+func (s implBlobManagerFetchBlobServerCallSend) Send(item BlobFetchStatus) error {
 	return s.s.Send(item)
 }
 
diff --git a/v23/services/syncbase/nosql/types.vdl b/v23/services/syncbase/nosql/types.vdl
index 37fd383..eae8e9d 100644
--- a/v23/services/syncbase/nosql/types.vdl
+++ b/v23/services/syncbase/nosql/types.vdl
@@ -156,19 +156,19 @@
 	NullBlobRef = BlobRef("")
 )
 
-// FetchState represents the state transitions of a blob fetch.
-type FetchState enum {
+// BlobFetchState represents the state transitions of a blob fetch.
+type BlobFetchState enum {
 	Pending // Fetch request is queued.
 	Locating // Blob discovery is in progress to find a source for the blob.
 	Fetching // Blob transfer is in progress.
 	Done // Blob is locally cached.
 }
 
-// FetchStatus describes the progress of an asynchronous blob fetch.
-type FetchStatus struct {
-	State FetchState // State of the blob fetch request.
-	Received uint64 // Total number of bytes received.
-	Total uint64 // Blob size.
+// BlobFetchStatus describes the progress of an asynchronous blob fetch.
+type BlobFetchStatus struct {
+	State BlobFetchState // State of the blob fetch request.
+	Received int64 // Total number of bytes received.
+	Total int64 // Blob size.
 }
 
 // StoreChange is the new value for a watched entity.
diff --git a/v23/services/syncbase/nosql/types.vdl.go b/v23/services/syncbase/nosql/types.vdl.go
index ac27aad..adb4e66 100644
--- a/v23/services/syncbase/nosql/types.vdl.go
+++ b/v23/services/syncbase/nosql/types.vdl.go
@@ -224,75 +224,75 @@
 }) {
 }
 
-// FetchState represents the state transitions of a blob fetch.
-type FetchState int
+// BlobFetchState represents the state transitions of a blob fetch.
+type BlobFetchState int
 
 const (
-	FetchStatePending FetchState = iota
-	FetchStateLocating
-	FetchStateFetching
-	FetchStateDone
+	BlobFetchStatePending BlobFetchState = iota
+	BlobFetchStateLocating
+	BlobFetchStateFetching
+	BlobFetchStateDone
 )
 
-// FetchStateAll holds all labels for FetchState.
-var FetchStateAll = [...]FetchState{FetchStatePending, FetchStateLocating, FetchStateFetching, FetchStateDone}
+// BlobFetchStateAll holds all labels for BlobFetchState.
+var BlobFetchStateAll = [...]BlobFetchState{BlobFetchStatePending, BlobFetchStateLocating, BlobFetchStateFetching, BlobFetchStateDone}
 
-// FetchStateFromString creates a FetchState from a string label.
-func FetchStateFromString(label string) (x FetchState, err error) {
+// BlobFetchStateFromString creates a BlobFetchState from a string label.
+func BlobFetchStateFromString(label string) (x BlobFetchState, err error) {
 	err = x.Set(label)
 	return
 }
 
 // Set assigns label to x.
-func (x *FetchState) Set(label string) error {
+func (x *BlobFetchState) Set(label string) error {
 	switch label {
 	case "Pending", "pending":
-		*x = FetchStatePending
+		*x = BlobFetchStatePending
 		return nil
 	case "Locating", "locating":
-		*x = FetchStateLocating
+		*x = BlobFetchStateLocating
 		return nil
 	case "Fetching", "fetching":
-		*x = FetchStateFetching
+		*x = BlobFetchStateFetching
 		return nil
 	case "Done", "done":
-		*x = FetchStateDone
+		*x = BlobFetchStateDone
 		return nil
 	}
 	*x = -1
-	return fmt.Errorf("unknown label %q in nosql.FetchState", label)
+	return fmt.Errorf("unknown label %q in nosql.BlobFetchState", label)
 }
 
 // String returns the string label of x.
-func (x FetchState) String() string {
+func (x BlobFetchState) String() string {
 	switch x {
-	case FetchStatePending:
+	case BlobFetchStatePending:
 		return "Pending"
-	case FetchStateLocating:
+	case BlobFetchStateLocating:
 		return "Locating"
-	case FetchStateFetching:
+	case BlobFetchStateFetching:
 		return "Fetching"
-	case FetchStateDone:
+	case BlobFetchStateDone:
 		return "Done"
 	}
 	return ""
 }
 
-func (FetchState) __VDLReflect(struct {
-	Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.FetchState"`
+func (BlobFetchState) __VDLReflect(struct {
+	Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.BlobFetchState"`
 	Enum struct{ Pending, Locating, Fetching, Done string }
 }) {
 }
 
-// FetchStatus describes the progress of an asynchronous blob fetch.
-type FetchStatus struct {
-	State    FetchState // State of the blob fetch request.
-	Received uint64     // Total number of bytes received.
-	Total    uint64     // Blob size.
+// BlobFetchStatus describes the progress of an asynchronous blob fetch.
+type BlobFetchStatus struct {
+	State    BlobFetchState // State of the blob fetch request.
+	Received int64          // Total number of bytes received.
+	Total    int64          // Blob size.
 }
 
-func (FetchStatus) __VDLReflect(struct {
-	Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.FetchStatus"`
+func (BlobFetchStatus) __VDLReflect(struct {
+	Name string `vdl:"v.io/syncbase/v23/services/syncbase/nosql.BlobFetchStatus"`
 }) {
 }
 
@@ -323,8 +323,8 @@
 	vdl.Register((*CrPolicy)(nil))
 	vdl.Register((*CrRule)(nil))
 	vdl.Register((*BlobRef)(nil))
-	vdl.Register((*FetchState)(nil))
-	vdl.Register((*FetchStatus)(nil))
+	vdl.Register((*BlobFetchState)(nil))
+	vdl.Register((*BlobFetchStatus)(nil))
 	vdl.Register((*StoreChange)(nil))
 }
 
diff --git a/v23/syncbase/nosql/blob.go b/v23/syncbase/nosql/blob.go
new file mode 100644
index 0000000..37cddc4
--- /dev/null
+++ b/v23/syncbase/nosql/blob.go
@@ -0,0 +1,294 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package nosql
+
+import (
+	"sync"
+
+	wire "v.io/syncbase/v23/services/syncbase/nosql"
+	"v.io/v23/context"
+	"v.io/v23/verror"
+)
+
+var _ Blob = (*blob)(nil)
+
+type blob struct {
+	c  wire.DatabaseClientMethods
+	br wire.BlobRef
+}
+
+func newBlob(dbName string, br wire.BlobRef) Blob {
+	return &blob{
+		c:  wire.DatabaseClient(dbName),
+		br: br,
+	}
+}
+
+func createBlob(ctx *context.T, dbName string) (Blob, error) {
+	b := &blob{
+		c: wire.DatabaseClient(dbName),
+	}
+	var err error
+	b.br, err = b.c.CreateBlob(ctx)
+	return b, err
+}
+
+// Ref implements Blob.Ref.
+func (b *blob) Ref() wire.BlobRef {
+	return b.br
+}
+
+// Put implements Blob.Put.
+func (b *blob) Put(ctx *context.T) (BlobWriter, error) {
+	call, err := b.c.PutBlob(ctx, b.br)
+	if err != nil {
+		return nil, err
+	}
+	return newBlobWriter(call), nil
+}
+
+// Commit implements Blob.Commit.
+func (b *blob) Commit(ctx *context.T) error {
+	return b.c.CommitBlob(ctx, b.br)
+}
+
+// Size implements Blob.Size.
+func (b *blob) Size(ctx *context.T) (int64, error) {
+	return b.c.GetBlobSize(ctx, b.br)
+}
+
+// Delete implements Blob.Delete.
+func (b *blob) Delete(ctx *context.T) error {
+	return b.c.DeleteBlob(ctx, b.br)
+}
+
+// Get implements Blob.Get.
+func (b *blob) Get(ctx *context.T, offset int64) (BlobReader, error) {
+	ctx, cancel := context.WithCancel(ctx)
+	call, err := b.c.GetBlob(ctx, b.br, offset)
+	if err != nil {
+		cancel()
+		return nil, err
+	}
+	return newBlobReader(cancel, call), nil
+}
+
+// Fetch implements Blob.Fetch.
+func (b *blob) Fetch(ctx *context.T, priority uint64) (BlobStatus, error) {
+	ctx, cancel := context.WithCancel(ctx)
+	call, err := b.c.FetchBlob(ctx, b.br, priority)
+	if err != nil {
+		cancel()
+		return nil, err
+	}
+	return newBlobStatus(cancel, call), nil
+}
+
+// Pin implements Blob.Pin.
+func (b *blob) Pin(ctx *context.T) error {
+	return b.c.PinBlob(ctx, b.br)
+}
+
+// Unpin implements Blob.Unpin.
+func (b *blob) Unpin(ctx *context.T) error {
+	return b.c.UnpinBlob(ctx, b.br)
+}
+
+// Keep implements Blob.Keep.
+func (b *blob) Keep(ctx *context.T, rank uint64) error {
+	return b.c.KeepBlob(ctx, b.br, rank)
+}
+
+////////////////////////////////////////
+// BlobWriter methods.
+
+type blobWriter struct {
+	mu sync.Mutex
+	// call is the RPC stream object.
+	call wire.BlobManagerPutBlobClientCall
+	// finished records whether we have called call.Finish().
+	finished bool
+}
+
+var _ BlobWriter = (*blobWriter)(nil)
+
+func newBlobWriter(call wire.BlobManagerPutBlobClientCall) *blobWriter {
+	return &blobWriter{
+		call: call,
+	}
+}
+
+// Send implements BlobWriter.Send.
+func (bw *blobWriter) Send(buf []byte) error {
+	return bw.call.SendStream().Send(buf)
+}
+
+// Close implements BlobWriter.Close.
+func (bw *blobWriter) Close() error {
+	bw.mu.Lock()
+	defer bw.mu.Unlock()
+	if !bw.finished {
+		// No need to call Close explicitly. Finish will take care of
+		// that.
+		bw.finished = true
+		return bw.call.Finish()
+	}
+	return nil
+}
+
+////////////////////////////////////////
+// BlobReader methods.
+// (similar to methods in stream.go).
+
+type blobReader struct {
+	mu sync.Mutex
+	// cancel cancels the RPC stream.
+	cancel context.CancelFunc
+	// call is the RPC stream object.
+	call wire.BlobManagerGetBlobClientCall
+	// curr is the currently staged bytes, or nil if nothing is staged.
+	curr []byte
+	// err is the first error encountered during streaming. It may also be
+	// populated by a call to Cancel.
+	err error
+	// finished records whether we have called call.Finish().
+	finished bool
+}
+
+var _ BlobReader = (*blobReader)(nil)
+
+func newBlobReader(cancel context.CancelFunc, call wire.BlobManagerGetBlobClientCall) *blobReader {
+	return &blobReader{
+		cancel: cancel,
+		call:   call,
+	}
+}
+
+// Advance implements BlobReader.Advance.
+func (br *blobReader) Advance() bool {
+	br.mu.Lock()
+	defer br.mu.Unlock()
+	if br.err != nil || br.finished {
+		return false
+	}
+	if br.call.RecvStream().Advance() {
+		br.curr = br.call.RecvStream().Value()
+		return true
+	}
+
+	br.err = br.call.RecvStream().Err()
+	err := br.call.Finish()
+	if br.err == nil {
+		br.err = err
+	}
+	br.cancel()
+	br.finished = true
+	return false
+}
+
+// Value implements BlobReader.Value.
+func (br *blobReader) Value() []byte {
+	br.mu.Lock()
+	defer br.mu.Unlock()
+	if br.curr == nil {
+		panic("nothing staged")
+	}
+	return br.curr
+}
+
+// Err implements BlobReader.Err.
+func (br *blobReader) Err() error {
+	br.mu.Lock()
+	defer br.mu.Unlock()
+	return br.err
+}
+
+// Cancel implements BlobReader.Cancel.
+// TODO(hpucha): Make Cancel non-blocking. Copied from stream.go
+func (br *blobReader) Cancel() {
+	br.mu.Lock()
+	defer br.mu.Unlock()
+	br.cancel()
+	br.call.Finish()
+	br.err = verror.New(verror.ErrCanceled, nil)
+}
+
+////////////////////////////////////////
+// BlobStatus methods.
+// (similar to methods in stream.go).
+
+type blobStatus struct {
+	mu sync.Mutex
+	// cancel cancels the RPC stream.
+	cancel context.CancelFunc
+	// call is the RPC stream object.
+	call wire.BlobManagerFetchBlobClientCall
+	// curr is the currently staged item, or nil if nothing is staged.
+	curr *wire.BlobFetchStatus
+	// err is the first error encountered during streaming. It may also be
+	// populated by a call to Cancel.
+	err error
+	// finished records whether we have called call.Finish().
+	finished bool
+}
+
+var _ BlobStatus = (*blobStatus)(nil)
+
+func newBlobStatus(cancel context.CancelFunc, call wire.BlobManagerFetchBlobClientCall) *blobStatus {
+	return &blobStatus{
+		cancel: cancel,
+		call:   call,
+	}
+}
+
+// Advance implements BlobStatus.Advance.
+func (bs *blobStatus) Advance() bool {
+	bs.mu.Lock()
+	defer bs.mu.Unlock()
+	if bs.err != nil || bs.finished {
+		return false
+	}
+	if bs.call.RecvStream().Advance() {
+		val := bs.call.RecvStream().Value()
+		bs.curr = &val
+		return true
+	}
+
+	bs.err = bs.call.RecvStream().Err()
+	err := bs.call.Finish()
+	if bs.err == nil {
+		bs.err = err
+	}
+	bs.cancel()
+	bs.finished = true
+	return false
+}
+
+// Value implements BlobStatus.Value.
+func (bs *blobStatus) Value() wire.BlobFetchStatus {
+	bs.mu.Lock()
+	defer bs.mu.Unlock()
+	if bs.curr == nil {
+		panic("nothing staged")
+	}
+	return *bs.curr
+}
+
+// Err implements BlobStatus.Err.
+func (bs *blobStatus) Err() error {
+	bs.mu.Lock()
+	defer bs.mu.Unlock()
+	return bs.err
+}
+
+// Cancel implements BlobStatus.Cancel.
+// TODO(hpucha): Make Cancel non-blocking. Copied from stream.go
+func (bs *blobStatus) Cancel() {
+	bs.mu.Lock()
+	defer bs.mu.Unlock()
+	bs.cancel()
+	bs.call.Finish()
+	bs.err = verror.New(verror.ErrCanceled, nil)
+}
diff --git a/v23/syncbase/nosql/blob_test.go b/v23/syncbase/nosql/blob_test.go
new file mode 100644
index 0000000..17cd2a8
--- /dev/null
+++ b/v23/syncbase/nosql/blob_test.go
@@ -0,0 +1,135 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package nosql_test
+
+import (
+	"reflect"
+	"testing"
+
+	wire "v.io/syncbase/v23/services/syncbase/nosql"
+	"v.io/syncbase/v23/syncbase"
+	"v.io/syncbase/v23/syncbase/nosql"
+	tu "v.io/syncbase/v23/syncbase/testutil"
+	"v.io/v23/context"
+	"v.io/v23/verror"
+)
+
+// Tests local blob get following a put.
+func TestLocalBlobPutGet(t *testing.T) {
+	ctx, sName, cleanup := tu.SetupOrDie(perms("root/client"))
+	defer cleanup()
+	a := tu.CreateApp(t, ctx, syncbase.NewService(sName), "a")
+	d := tu.CreateNoSQLDatabase(t, ctx, a, "d")
+
+	b, err := d.CreateBlob(ctx)
+	if err != nil {
+		t.Fatalf("Creating local blob failed, err %v", err)
+	}
+
+	// Verify that getting/fetching an uncommitted blob should fail.
+	br, err := b.Get(ctx, 0)
+	if err != nil {
+		t.Fatalf("GetBlob RPC failed, err %v", err)
+	}
+	if br.Advance() {
+		t.Fatalf("Getting uncommitted blob didn't fail")
+	}
+	if verror.ErrorID(br.Err()) != wire.ErrBlobNotCommitted.ID {
+		t.Fatalf("Getting uncommitted blob didn't fail, err %v", err)
+	}
+
+	bs, err := b.Fetch(ctx, 100)
+	if err != nil {
+		t.Fatalf("FetchBlob RPC failed, err %v", err)
+	}
+	if bs.Advance() {
+		t.Fatalf("Fetching uncommitted blob didn't fail")
+	}
+	if verror.ErrorID(bs.Err()) != wire.ErrBlobNotCommitted.ID {
+		t.Fatalf("Fetching uncommitted blob didn't fail, err %v", err)
+	}
+
+	verifyBlobSize(t, ctx, b, 0)
+
+	// Put some data in the blob.
+	bw, err := b.Put(ctx)
+	if err != nil {
+		t.Fatalf("PutBlob RPC failed, err %v", err)
+	}
+	wantVal := "foobarbaz"
+	wantSize := int64(len(wantVal))
+	if err := bw.Send([]byte(wantVal)); err != nil {
+		t.Fatalf("Sending blob data failed, err %v", err)
+	}
+
+	// Before the put is completed, the blob size can be anything between 0
+	// and wantSize since the blob size is obtained from the persistent
+	// storage before closing. Note: if Fatalf is used when there is an
+	// error, the test hangs. Not sure why but looks like it is because
+	// there is still a pending goroutine spawned for the Put rpc.
+	gotSize, err := b.Size(ctx)
+	if err != nil || gotSize > wantSize {
+		t.Errorf("Getting blob size failed, got %v, want %v, err %v", gotSize, wantSize, err)
+	}
+
+	if err := bw.Close(); err != nil {
+		t.Fatalf("Closing blob writer failed, err %v", err)
+	}
+
+	verifyBlobSize(t, ctx, b, wantSize)
+
+	// Commit the blob.
+	if err := b.Commit(ctx); err != nil {
+		t.Fatalf("Committing a blob failed, err %v", err)
+	}
+
+	// Verify getting/fetching the committed blob.
+	verifyBlobSize(t, ctx, b, wantSize)
+	verifyGetBlob(t, ctx, b, 0, wantVal)
+	verifyGetBlob(t, ctx, b, 3, "barbaz")
+
+	bs, err = b.Fetch(ctx, 100)
+	if err != nil {
+		t.Fatalf("FetchBlob RPC failed, err %v", err)
+	}
+	wantStatus := wire.BlobFetchStatus{State: wire.BlobFetchStateDone}
+	var gotStatus wire.BlobFetchStatus
+	for bs.Advance() {
+		gotStatus = bs.Value()
+	}
+	if bs.Err() != nil {
+		t.Fatalf("Fetching a blob failed, err %v", br.Err())
+	}
+	if !reflect.DeepEqual(gotStatus, wantStatus) {
+		t.Fatalf("Fetching a blob failed, got %v want %v", gotStatus, wantStatus)
+	}
+}
+
+///////////////////
+// Helpers.
+
+func verifyBlobSize(t *testing.T, ctx *context.T, b nosql.Blob, wantSize int64) {
+	gotSize, err := b.Size(ctx)
+	if err != nil || gotSize != wantSize {
+		t.Fatalf("Getting blob size failed, got %v, want %v, err %v", gotSize, wantSize, err)
+	}
+}
+
+func verifyGetBlob(t *testing.T, ctx *context.T, b nosql.Blob, offset int64, wantVal string) {
+	br, err := b.Get(ctx, offset)
+	if err != nil {
+		t.Fatalf("GetBlob RPC failed, err %v", err)
+	}
+	var gotVal []byte
+	for br.Advance() {
+		gotVal = append(gotVal, br.Value()...)
+	}
+	if br.Err() != nil {
+		t.Fatalf("Getting a blob failed, err %v", br.Err())
+	}
+	if !reflect.DeepEqual(gotVal, []byte(wantVal)) {
+		t.Fatalf("Getting a blob failed, got %v want %v", gotVal, []byte(wantVal))
+	}
+}
diff --git a/v23/syncbase/nosql/blob_v23_test.go b/v23/syncbase/nosql/blob_v23_test.go
new file mode 100644
index 0000000..3abed9c
--- /dev/null
+++ b/v23/syncbase/nosql/blob_v23_test.go
@@ -0,0 +1,371 @@
+// Copyright 2015 The Vanadium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package nosql_test
+
+import (
+	"crypto/md5"
+	"crypto/rand"
+	"fmt"
+	"reflect"
+	"strconv"
+	"time"
+
+	wire "v.io/syncbase/v23/services/syncbase/nosql"
+	"v.io/syncbase/v23/syncbase"
+	tu "v.io/syncbase/v23/syncbase/testutil"
+	constants "v.io/syncbase/x/ref/services/syncbase/server/util"
+	"v.io/v23"
+	"v.io/v23/naming"
+	_ "v.io/x/ref/runtime/factories/generic"
+	"v.io/x/ref/test/modules"
+	"v.io/x/ref/test/v23tests"
+)
+
+//go:generate v23 test generate
+
+func V23TestSyncbasedWholeBlobTransfer(t *v23tests.T) {
+	v23tests.RunRootMT(t, "--v23.tcp.address=127.0.0.1:0")
+	server0Creds, _ := t.Shell().NewChildCredentials("s0")
+	client0Creds, _ := t.Shell().NewChildCredentials("c0")
+	cleanSync0 := tu.StartSyncbased(t, server0Creds, "sync0", "",
+		`{"Read": {"In":["root/c0"]}, "Write": {"In":["root/c0"]}}`)
+	defer cleanSync0()
+
+	server1Creds, _ := t.Shell().NewChildCredentials("s1")
+	client1Creds, _ := t.Shell().NewChildCredentials("c1")
+	cleanSync1 := tu.StartSyncbased(t, server1Creds, "sync1", "",
+		`{"Read": {"In":["root/c1"]}, "Write": {"In":["root/c1"]}}`)
+	defer cleanSync1()
+
+	sgName := naming.Join("sync0", constants.SyncbaseSuffix, "SG1")
+
+	tu.RunClient(t, client0Creds, runSetupAppA, "sync0")
+	tu.RunClient(t, client0Creds, runPopulateData, "sync0", "foo", "0")
+	tu.RunClient(t, client0Creds, runCreateSyncGroup, "sync0", sgName, "tb:foo", "root/s0", "root/s1")
+
+	tu.RunClient(t, client1Creds, runSetupAppA, "sync1")
+	tu.RunClient(t, client1Creds, runJoinSyncGroup, "sync1", sgName)
+	tu.RunClient(t, client1Creds, runVerifySyncGroupData, "sync1", "foo", "0", "10")
+
+	// FetchBlob first.
+	tu.RunClient(t, client0Creds, runGenerateBlob, "sync0", "foo", "0", "foobarbaz")
+	tu.RunClient(t, client1Creds, runFetchBlob, "sync1", "foo", "0", "9", "false")
+	tu.RunClient(t, client1Creds, runGetBlob, "sync1", "foo", "0", "foobarbaz", "0")
+
+	// GetBlob directly.
+	tu.RunClient(t, client1Creds, runGenerateBlob, "sync1", "foo", "0", "abcdefghijklmn")
+	tu.RunClient(t, client0Creds, runGetBlob, "sync0", "foo", "0", "fghijklmn", "5")
+	tu.RunClient(t, client0Creds, runFetchBlob, "sync0", "foo", "0", "14", "true")
+
+	// Test with a big blob (1 MB).
+	tu.RunClient(t, client0Creds, runGenerateBigBlob, "sync0", "foo", "1")
+	tu.RunClient(t, client1Creds, runGetBigBlob, "sync1", "foo", "1")
+}
+
+////////////////////////////////////
+// Helpers.
+
+type testStruct struct {
+	Val  string
+	Blob wire.BlobRef
+}
+
+// Arguments: 0: syncbase name, 1: key prefix, 2: key position (1+2 is the
+// actual key), 3: blob data.
+var runGenerateBlob = modules.Register(func(env *modules.Env, args ...string) error {
+	ctx, shutdown := v23.Init()
+	defer shutdown()
+
+	a := syncbase.NewService(args[0]).App("a")
+	d := a.NoSQLDatabase("d", nil)
+
+	b, err := d.CreateBlob(ctx)
+	if err != nil {
+		return fmt.Errorf("CreateBlob failed, err %v\n", err)
+	}
+	bw, err := b.Put(ctx)
+	if err != nil {
+		return fmt.Errorf("PutBlob RPC failed, err %v\n", err)
+	}
+
+	data := args[3]
+	if err := bw.Send([]byte(data)); err != nil {
+		return fmt.Errorf("Sending blob data failed, err %v\n", err)
+	}
+	if err := bw.Close(); err != nil {
+		return fmt.Errorf("Closing blob writer failed, err %v\n", err)
+	}
+
+	// Commit the blob.
+	if err := b.Commit(ctx); err != nil {
+		return fmt.Errorf("Committing a blob failed, err %v\n", err)
+	}
+
+	// Put the BlobRef in a key.
+	tb := d.Table("tb")
+	pos, _ := strconv.ParseUint(args[2], 10, 64)
+
+	key := fmt.Sprintf("%s%d", args[1], pos)
+	r := tb.Row(key)
+	s := testStruct{Val: "testkey" + key, Blob: b.Ref()}
+	if err := r.Put(ctx, s); err != nil {
+		return fmt.Errorf("r.Put() failed: %v\n", err)
+	}
+
+	return nil
+}, "runGenerateBlob")
+
+// Arguments: 0: syncbase name, 1: key prefix, 2: key position (1+2 is the
+// actual key), 3: blob size, 4: skip incremental status checking.
+var runFetchBlob = modules.Register(func(env *modules.Env, args ...string) error {
+	ctx, shutdown := v23.Init()
+	defer shutdown()
+
+	a := syncbase.NewService(args[0]).App("a")
+	d := a.NoSQLDatabase("d", nil)
+
+	tb := d.Table("tb")
+	pos, _ := strconv.ParseUint(args[2], 10, 64)
+
+	key := fmt.Sprintf("%s%d", args[1], pos)
+	r := tb.Row(key)
+	var s testStruct
+
+	// Try for 10 seconds to get the new value.
+	var err error
+	for i := 0; i < 10; i++ {
+		// Note: the error is a decode error since the old value is a
+		// string, and the new value is testStruct.
+		if err = r.Get(ctx, &s); err == nil {
+			break
+		}
+		time.Sleep(1 * time.Second)
+	}
+
+	if err != nil {
+		return fmt.Errorf("r.Get() failed: %v\n", err)
+	}
+
+	b := d.Blob(s.Blob)
+	bs, err := b.Fetch(ctx, 100)
+	if err != nil {
+		return fmt.Errorf("Fetch RPC failed, err %v\n", err)
+	}
+
+	status := []wire.BlobFetchStatus{
+		wire.BlobFetchStatus{State: wire.BlobFetchStatePending},
+		wire.BlobFetchStatus{State: wire.BlobFetchStateLocating},
+		wire.BlobFetchStatus{State: wire.BlobFetchStateFetching},
+		wire.BlobFetchStatus{State: wire.BlobFetchStateDone}}
+
+	skipIncStatus, _ := strconv.ParseBool(args[4])
+
+	var gotStatus wire.BlobFetchStatus
+	i := 0
+	for bs.Advance() {
+		gotStatus = bs.Value()
+
+		if !skipIncStatus {
+			if i <= 1 {
+				if !reflect.DeepEqual(gotStatus, status[i]) {
+					return fmt.Errorf("Fetch blob failed, got status %v want status %v\n", gotStatus, status[i])
+				}
+				i++
+			} else if !(gotStatus.State == status[2].State || reflect.DeepEqual(gotStatus, status[3])) {
+				return fmt.Errorf("Fetch blob failed, got status %v\n", gotStatus)
+			}
+		}
+	}
+
+	if !reflect.DeepEqual(gotStatus, status[3]) {
+		return fmt.Errorf("Fetch blob failed, got status %v want status %v\n", gotStatus, status[3])
+	}
+
+	if bs.Err() != nil {
+		return fmt.Errorf("Fetch blob failed, err %v\n", err)
+	}
+
+	wantSize, _ := strconv.ParseInt(args[3], 10, 64)
+	gotSize, err := b.Size(ctx)
+	if err != nil || gotSize != wantSize {
+		return fmt.Errorf("Blob size incorrect, got %v want %v, err %v\n", gotSize, wantSize, err)
+	}
+
+	return nil
+}, "runFetchBlob")
+
+// Arguments: 0: syncbase name, 1: key prefix, 2: key position (1+2 is the
+// actual key), 3: expected blob data, 4: offset for get.
+var runGetBlob = modules.Register(func(env *modules.Env, args ...string) error {
+	ctx, shutdown := v23.Init()
+	defer shutdown()
+
+	a := syncbase.NewService(args[0]).App("a")
+	d := a.NoSQLDatabase("d", nil)
+
+	tb := d.Table("tb")
+	pos, _ := strconv.ParseUint(args[2], 10, 64)
+
+	key := fmt.Sprintf("%s%d", args[1], pos)
+	r := tb.Row(key)
+	var s testStruct
+
+	// Try for 10 seconds to get the new value.
+	var err error
+	for i := 0; i < 10; i++ {
+		// Note: the error is a decode error since the old value is a
+		// string, and the new value is testStruct.
+		if err = r.Get(ctx, &s); err == nil {
+			break
+		}
+		time.Sleep(1 * time.Second)
+	}
+
+	if err != nil {
+		return fmt.Errorf("r.Get() failed: %v\n", err)
+	}
+
+	b := d.Blob(s.Blob)
+	offset, _ := strconv.ParseInt(args[4], 10, 64)
+	br, err := b.Get(ctx, offset)
+	if err != nil {
+		return fmt.Errorf("GetBlob RPC failed, err %v\n", err)
+	}
+	var gotVal []byte
+	for br.Advance() {
+		gotVal = append(gotVal, br.Value()...)
+	}
+	if br.Err() != nil {
+		return fmt.Errorf("Getting a blob failed, err %v\n", br.Err())
+	}
+	if !reflect.DeepEqual(gotVal, []byte(args[3])) {
+		return fmt.Errorf("Getting a blob failed, got %v want %v\n", gotVal, []byte(args[3]))
+	}
+
+	return nil
+}, "runGetBlob")
+
+// Arguments: 0: syncbase name, 1: key prefix, 2: key position (1+2 is the
+// actual key).
+var runGenerateBigBlob = modules.Register(func(env *modules.Env, args ...string) error {
+	ctx, shutdown := v23.Init()
+	defer shutdown()
+
+	a := syncbase.NewService(args[0]).App("a")
+	d := a.NoSQLDatabase("d", nil)
+
+	b, err := d.CreateBlob(ctx)
+	if err != nil {
+		return fmt.Errorf("CreateBlob failed, err %v\n", err)
+	}
+	bw, err := b.Put(ctx)
+	if err != nil {
+		return fmt.Errorf("PutBlob RPC failed, err %v\n", err)
+	}
+
+	hasher := md5.New()
+
+	chunkSize := 8192
+	content := make([]byte, chunkSize)
+	// Send 1 MB blob.
+	for i := 0; i < 128; i++ {
+		if n, err := rand.Read(content); err != nil || n != chunkSize {
+			return fmt.Errorf("Creating blob data failed, n %v err %v\n", n, err)
+		}
+		if err := bw.Send(content); err != nil {
+			return fmt.Errorf("Sending blob data failed, err %v\n", err)
+		}
+		hasher.Write(content)
+	}
+	if err := bw.Close(); err != nil {
+		return fmt.Errorf("Closing blob writer failed, err %v\n", err)
+	}
+
+	// Commit the blob.
+	if err := b.Commit(ctx); err != nil {
+		return fmt.Errorf("Committing a blob failed, err %v\n", err)
+	}
+
+	// Put the BlobRef in a key.
+	tb := d.Table("tb")
+	pos, _ := strconv.ParseUint(args[2], 10, 64)
+
+	key := fmt.Sprintf("%s%d", args[1], pos)
+	r := tb.Row(key)
+
+	// Blob hash is transferred via structured store.
+	s := testStruct{Val: hashToString(hasher.Sum(nil)), Blob: b.Ref()}
+
+	if err := r.Put(ctx, s); err != nil {
+		return fmt.Errorf("r.Put() failed: %v\n", err)
+	}
+
+	return nil
+}, "runGenerateBigBlob")
+
+// Arguments: 0: syncbase name, 1: key prefix, 2: key position (1+2 is the
+// actual key).
+var runGetBigBlob = modules.Register(func(env *modules.Env, args ...string) error {
+	ctx, shutdown := v23.Init()
+	defer shutdown()
+
+	a := syncbase.NewService(args[0]).App("a")
+	d := a.NoSQLDatabase("d", nil)
+
+	tb := d.Table("tb")
+	pos, _ := strconv.ParseUint(args[2], 10, 64)
+
+	key := fmt.Sprintf("%s%d", args[1], pos)
+	r := tb.Row(key)
+	var s testStruct
+
+	// Try for 10 seconds to get the new value.
+	var err error
+	for i := 0; i < 10; i++ {
+		// Note: the error is a decode error since the old value is a
+		// string, and the new value is testStruct.
+		if err = r.Get(ctx, &s); err == nil {
+			break
+		}
+		time.Sleep(1 * time.Second)
+	}
+
+	if err != nil {
+		return fmt.Errorf("r.Get() failed: %v\n", err)
+	}
+
+	b := d.Blob(s.Blob)
+	br, err := b.Get(ctx, 0)
+	if err != nil {
+		return fmt.Errorf("GetBlob RPC failed, err %v\n", err)
+	}
+	hasher := md5.New()
+	for br.Advance() {
+		content := br.Value()
+		hasher.Write(content)
+	}
+	if br.Err() != nil {
+		return fmt.Errorf("Getting a blob failed, err %v\n", br.Err())
+	}
+
+	gotHash := hashToString(hasher.Sum(nil))
+	if !reflect.DeepEqual(gotHash, s.Val) {
+		return fmt.Errorf("Getting a blob failed, got %v want %v\n", gotHash, s.Val)
+	}
+
+	return nil
+}, "runGetBigBlob")
+
+// Copied from localblobstore/fs_cablobstore/fs_cablobstore.go.
+//
+// hashToString() returns a string representation of the hash.
+// Requires len(hash)==16.  An md5 hash is suitable.
+func hashToString(hash []byte) string {
+	return fmt.Sprintf("%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x",
+		hash[0], hash[1], hash[2], hash[3],
+		hash[4], hash[5], hash[6], hash[7],
+		hash[8], hash[9], hash[10], hash[11],
+		hash[12], hash[13], hash[14], hash[15])
+}
diff --git a/v23/syncbase/nosql/database.go b/v23/syncbase/nosql/database.go
index 5a23295..5e8159d 100644
--- a/v23/syncbase/nosql/database.go
+++ b/v23/syncbase/nosql/database.go
@@ -139,6 +139,16 @@
 	return d.c.GetSyncGroupNames(ctx)
 }
 
+// Blob implements Database.Blob.
+func (d *database) Blob(br wire.BlobRef) Blob {
+	return newBlob(d.fullName, br)
+}
+
+// CreateBlob implements Database.CreateBlob.
+func (d *database) CreateBlob(ctx *context.T) (Blob, error) {
+	return createBlob(ctx, d.fullName)
+}
+
 // UpgradeIfOutdated implements Database.UpgradeIfOutdated.
 func (d *database) UpgradeIfOutdated(ctx *context.T) (bool, error) {
 	var schema *Schema = d.schema
diff --git a/v23/syncbase/nosql/model.go b/v23/syncbase/nosql/model.go
index 99f11eb..e26b76d 100644
--- a/v23/syncbase/nosql/model.go
+++ b/v23/syncbase/nosql/model.go
@@ -105,6 +105,12 @@
 	// this database.
 	GetSyncGroupNames(ctx *context.T) ([]string, error)
 
+	// CreateBlob returns a handle to the new Blob instantiated by Syncbase.
+	CreateBlob(ctx *context.T) (Blob, error)
+
+	// Blob returns a handle to the blob with the given BlobRef.
+	Blob(br wire.BlobRef) Blob
+
 	// This method compares the current schema version of the database with the
 	// schema version provided while creating this database handle. If the
 	// current database schema version is lower, then the SchemaUpdater is
@@ -342,6 +348,104 @@
 	GetMembers(ctx *context.T) (map[string]wire.SyncGroupMemberInfo, error)
 }
 
+// Blob is the interface for a Blob in the store.
+type Blob interface {
+	// Ref returns Syncbase's BlobRef for this blob.
+	Ref() wire.BlobRef
+
+	// Put appends the byte stream to the blob.
+	Put(ctx *context.T) (BlobWriter, error)
+
+	// Commit marks the blob as immutable.
+	Commit(ctx *context.T) error
+
+	// Size returns the count of bytes written as part of the blob
+	// (committed or uncommitted).
+	Size(ctx *context.T) (int64, error)
+
+	// Delete locally deletes the blob (committed or uncommitted).
+	Delete(ctx *context.T) error
+
+	// Get returns the byte stream from a committed blob starting at offset.
+	Get(ctx *context.T, offset int64) (BlobReader, error)
+
+	// Fetch initiates fetching a blob if not locally found. priority
+	// controls the network priority of the blob. Higher priority blobs are
+	// fetched before the lower priority ones. However an ongoing blob
+	// transfer is not interrupted. Status updates are streamed back to the
+	// client as fetch is in progress.
+	Fetch(ctx *context.T, priority uint64) (BlobStatus, error)
+
+	// Pin locally pins the blob so that it is not evicted.
+	Pin(ctx *context.T) error
+
+	// Unpin locally unpins the blob so that it can be evicted if needed.
+	Unpin(ctx *context.T) error
+
+	// Keep locally caches the blob with the specified rank. Lower
+	// ranked blobs are more eagerly evicted.
+	Keep(ctx *context.T, rank uint64) error
+}
+
+// BlobWriter is an interface for putting a blob.
+type BlobWriter interface {
+	// Send places the bytes given by the client onto the output
+	// stream. Returns errors encountered while sending. Blocks if there is
+	// no buffer space.
+	Send([]byte) error
+
+	// Close indicates that no more bytes will be sent.
+	Close() error
+}
+
+// BlobReader is an interface for getting a blob.
+type BlobReader interface {
+	// Advance() stages bytes so that they may be retrieved via
+	// Value(). Returns true iff there are bytes to retrieve. Advance() must
+	// be called before Value() is called. The caller is expected to read
+	// until Advance() returns false, or to call Cancel().
+	Advance() bool
+
+	// Value() returns the bytes that were staged by Advance(). May panic if
+	// Advance() returned false or was not called. Never blocks.
+	Value() []byte
+
+	// Err() returns any error encountered by Advance. Never blocks.
+	Err() error
+
+	// Cancel notifies the stream provider that it can stop producing
+	// elements.  The client must call Cancel if it does not iterate through
+	// all elements (i.e. until Advance returns false). Cancel is idempotent
+	// and can be called concurrently with a goroutine that is iterating via
+	// Advance.  Cancel causes Advance to subsequently return false. Cancel
+	// does not block.
+	Cancel()
+}
+
+// BlobStatus is an interface for getting the status of a blob transfer.
+type BlobStatus interface {
+	// Advance() stages an item so that it may be retrieved via
+	// Value(). Returns true iff there are items to retrieve. Advance() must
+	// be called before Value() is called. The caller is expected to read
+	// until Advance() returns false, or to call Cancel().
+	Advance() bool
+
+	// Value() returns the item that was staged by Advance(). May panic if
+	// Advance() returned false or was not called. Never blocks.
+	Value() wire.BlobFetchStatus
+
+	// Err() returns any error encountered by Advance. Never blocks.
+	Err() error
+
+	// Cancel notifies the stream provider that it can stop producing
+	// elements.  The client must call Cancel if it does not iterate through
+	// all elements (i.e. until Advance returns false). Cancel is idempotent
+	// and can be called concurrently with a goroutine that is iterating via
+	// Advance.  Cancel causes Advance to subsequently return false. Cancel
+	// does not block.
+	Cancel()
+}
+
 // SchemaUpgrader interface must be implemented by the App in order to upgrade
 // the database schema from a lower version to a higher version.
 type SchemaUpgrader interface {
diff --git a/v23/syncbase/nosql/v23_test.go b/v23/syncbase/nosql/v23_test.go
index d235b1b..39da63c 100644
--- a/v23/syncbase/nosql/v23_test.go
+++ b/v23/syncbase/nosql/v23_test.go
@@ -25,6 +25,10 @@
 	os.Exit(r)
 }
 
+func TestV23SyncbasedWholeBlobTransfer(t *testing.T) {
+	v23tests.RunTest(t, V23TestSyncbasedWholeBlobTransfer)
+}
+
 func TestV23SyncbasedJoinSyncGroup(t *testing.T) {
 	v23tests.RunTest(t, V23TestSyncbasedJoinSyncGroup)
 }
diff --git a/x/ref/services/syncbase/localblobstore/localblobstore_transfer_test.go b/x/ref/services/syncbase/localblobstore/localblobstore_transfer_test.go
index c3f11d4..b5378ec 100644
--- a/x/ref/services/syncbase/localblobstore/localblobstore_transfer_test.go
+++ b/x/ref/services/syncbase/localblobstore/localblobstore_transfer_test.go
@@ -258,7 +258,7 @@
 		rs := bsR.RecipeStreamFromChunkStream(ctx, cs)
 
 		// 6. The following thread sends the chunk hashes that the
-		//    receiver does not have are to the sender.  It also makes
+		//    receiver does not have to the sender.  It also makes
 		//    a duplicate of the stream on the channel rsCopy.  The
 		//    buffering in rsCopy allows the receiver to put several
 		//    chunks into a fragment.
diff --git a/x/ref/services/syncbase/localblobstore/model.go b/x/ref/services/syncbase/localblobstore/model.go
index ba08726..f51f455 100644
--- a/x/ref/services/syncbase/localblobstore/model.go
+++ b/x/ref/services/syncbase/localblobstore/model.go
@@ -97,7 +97,7 @@
 	// that this will be called on a receiving device, and be given a
 	// ChunkStream from a sending device, to yield a recipe for efficient
 	// chunk transfer.  RecipeStep values with non-nil Chunk fields need
-	// the chunk from the sender; once the data is returned is can be
+	// the chunk from the sender; once the data is returned it can be
 	// written with BlobWriter.AppendFragment().  Those with blob
 	// references can be written locally with BlobWriter.AppendBlob().
 	RecipeStreamFromChunkStream(ctx *context.T, chunkStream ChunkStream) RecipeStream
diff --git a/x/ref/services/syncbase/server/app.go b/x/ref/services/syncbase/server/app.go
index d7103e9..c404f77 100644
--- a/x/ref/services/syncbase/server/app.go
+++ b/x/ref/services/syncbase/server/app.go
@@ -246,9 +246,6 @@
 	if err := d.St().Close(); err != nil {
 		return err
 	}
-	if err := d.BlobSt().Close(); err != nil {
-		return err
-	}
 	if err := util.DestroyStore(a.s.opts.Engine, a.rootDirForDb(dbName)); err != nil {
 		return err
 	}
diff --git a/x/ref/services/syncbase/server/interfaces/database.go b/x/ref/services/syncbase/server/interfaces/database.go
index 02e232e..8be30d8 100644
--- a/x/ref/services/syncbase/server/interfaces/database.go
+++ b/x/ref/services/syncbase/server/interfaces/database.go
@@ -5,7 +5,6 @@
 package interfaces
 
 import (
-	"v.io/syncbase/x/ref/services/syncbase/localblobstore"
 	"v.io/syncbase/x/ref/services/syncbase/store"
 	"v.io/v23/context"
 	"v.io/v23/rpc"
@@ -17,9 +16,6 @@
 	// St returns the storage engine instance for this database.
 	St() store.Store
 
-	// BlobSt returns the blob storage engine instance for this database.
-	BlobSt() localblobstore.BlobStore
-
 	// App returns the app handle for this database.
 	App() App
 
diff --git a/x/ref/services/syncbase/server/interfaces/sync.vdl b/x/ref/services/syncbase/server/interfaces/sync.vdl
index ee15a16..b97e845 100644
--- a/x/ref/services/syncbase/server/interfaces/sync.vdl
+++ b/x/ref/services/syncbase/server/interfaces/sync.vdl
@@ -37,6 +37,23 @@
 	JoinSyncGroupAtAdmin(sgName, joinerName string, myInfo wire.SyncGroupMemberInfo) (SyncGroup | error) {access.Read}
 
 	// BlobSync methods.
-	// FetchBlob returns the requested blob.
-	FetchBlob(br wire.BlobRef) error {access.Read}
+
+	// HaveBlob verifies that the peer has the requested blob, and if
+	// present, returns its size.
+	HaveBlob(br wire.BlobRef) (int64 | error)
+
+	// FetchBlob fetches the requested blob.
+	FetchBlob(br wire.BlobRef) stream<_, []byte> error
+
+	// Methods for incremental blob transfer. The transfer starts with the
+	// receiver making a FetchBlobRecipe call to the sender for a given
+	// BlobRef. The sender, in turn, sends the chunk hashes of all the
+	// chunks that make up the requested blob (blob recipe). The receiver
+	// looks up the chunk hashes in its local blob store, and identifies the
+	// missing ones. The receiver then fetches the missing chunks using a
+	// FetchChunks call from the sender. Finally, the receiver finishes the
+	// blob fetch by combining the chunks obtained over the network with the
+	// already available local chunks as per the blob recipe.
+	FetchBlobRecipe(br wire.BlobRef) stream<_, ChunkHash> error
+	FetchChunks() stream<ChunkHash, ChunkData> error
 }
diff --git a/x/ref/services/syncbase/server/interfaces/sync.vdl.go b/x/ref/services/syncbase/server/interfaces/sync.vdl.go
index d378170..9cd1383 100644
--- a/x/ref/services/syncbase/server/interfaces/sync.vdl.go
+++ b/x/ref/services/syncbase/server/interfaces/sync.vdl.go
@@ -45,9 +45,22 @@
 	// allowed to join the named SyncGroup, and if so, adds the requestor to
 	// the SyncGroup.
 	JoinSyncGroupAtAdmin(ctx *context.T, sgName string, joinerName string, myInfo nosql.SyncGroupMemberInfo, opts ...rpc.CallOpt) (SyncGroup, error)
-	// BlobSync methods.
-	// FetchBlob returns the requested blob.
-	FetchBlob(ctx *context.T, br nosql.BlobRef, opts ...rpc.CallOpt) error
+	// HaveBlob verifies that the peer has the requested blob, and if
+	// present, returns its size.
+	HaveBlob(ctx *context.T, br nosql.BlobRef, opts ...rpc.CallOpt) (int64, error)
+	// FetchBlob fetches the requested blob.
+	FetchBlob(ctx *context.T, br nosql.BlobRef, opts ...rpc.CallOpt) (SyncFetchBlobClientCall, error)
+	// Methods for incremental blob transfer. The transfer starts with the
+	// receiver making a FetchBlobRecipe call to the sender for a given
+	// BlobRef. The sender, in turn, sends the chunk hashes of all the
+	// chunks that make up the requested blob (blob recipe). The receiver
+	// looks up the chunk hashes in its local blob store, and identifies the
+	// missing ones. The receiver then fetches the missing chunks using a
+	// FetchChunks call from the sender. Finally, the receiver finishes the
+	// blob fetch by combining the chunks obtained over the network with the
+	// already available local chunks as per the blob recipe.
+	FetchBlobRecipe(ctx *context.T, br nosql.BlobRef, opts ...rpc.CallOpt) (SyncFetchBlobRecipeClientCall, error)
+	FetchChunks(*context.T, ...rpc.CallOpt) (SyncFetchChunksClientCall, error)
 }
 
 // SyncClientStub adds universal methods to SyncClientMethods.
@@ -84,8 +97,35 @@
 	return
 }
 
-func (c implSyncClientStub) FetchBlob(ctx *context.T, i0 nosql.BlobRef, opts ...rpc.CallOpt) (err error) {
-	err = v23.GetClient(ctx).Call(ctx, c.name, "FetchBlob", []interface{}{i0}, nil, opts...)
+func (c implSyncClientStub) HaveBlob(ctx *context.T, i0 nosql.BlobRef, opts ...rpc.CallOpt) (o0 int64, err error) {
+	err = v23.GetClient(ctx).Call(ctx, c.name, "HaveBlob", []interface{}{i0}, []interface{}{&o0}, opts...)
+	return
+}
+
+func (c implSyncClientStub) FetchBlob(ctx *context.T, i0 nosql.BlobRef, opts ...rpc.CallOpt) (ocall SyncFetchBlobClientCall, err error) {
+	var call rpc.ClientCall
+	if call, err = v23.GetClient(ctx).StartCall(ctx, c.name, "FetchBlob", []interface{}{i0}, opts...); err != nil {
+		return
+	}
+	ocall = &implSyncFetchBlobClientCall{ClientCall: call}
+	return
+}
+
+func (c implSyncClientStub) FetchBlobRecipe(ctx *context.T, i0 nosql.BlobRef, opts ...rpc.CallOpt) (ocall SyncFetchBlobRecipeClientCall, err error) {
+	var call rpc.ClientCall
+	if call, err = v23.GetClient(ctx).StartCall(ctx, c.name, "FetchBlobRecipe", []interface{}{i0}, opts...); err != nil {
+		return
+	}
+	ocall = &implSyncFetchBlobRecipeClientCall{ClientCall: call}
+	return
+}
+
+func (c implSyncClientStub) FetchChunks(ctx *context.T, opts ...rpc.CallOpt) (ocall SyncFetchChunksClientCall, err error) {
+	var call rpc.ClientCall
+	if call, err = v23.GetClient(ctx).StartCall(ctx, c.name, "FetchChunks", nil, opts...); err != nil {
+		return
+	}
+	ocall = &implSyncFetchChunksClientCall{ClientCall: call}
 	return
 }
 
@@ -191,6 +231,246 @@
 	return
 }
 
+// SyncFetchBlobClientStream is the client stream for Sync.FetchBlob.
+type SyncFetchBlobClientStream interface {
+	// RecvStream returns the receiver side of the Sync.FetchBlob client stream.
+	RecvStream() interface {
+		// Advance stages an item so that it may be retrieved via Value.  Returns
+		// true iff there is an item to retrieve.  Advance must be called before
+		// Value is called.  May block if an item is not available.
+		Advance() bool
+		// Value returns the item that was staged by Advance.  May panic if Advance
+		// returned false or was not called.  Never blocks.
+		Value() []byte
+		// Err returns any error encountered by Advance.  Never blocks.
+		Err() error
+	}
+}
+
+// SyncFetchBlobClientCall represents the call returned from Sync.FetchBlob.
+type SyncFetchBlobClientCall interface {
+	SyncFetchBlobClientStream
+	// Finish blocks until the server is done, and returns the positional return
+	// values for call.
+	//
+	// Finish returns immediately if the call has been canceled; depending on the
+	// timing the output could either be an error signaling cancelation, or the
+	// valid positional return values from the server.
+	//
+	// Calling Finish is mandatory for releasing stream resources, unless the call
+	// has been canceled or any of the other methods return an error.  Finish should
+	// be called at most once.
+	Finish() error
+}
+
+type implSyncFetchBlobClientCall struct {
+	rpc.ClientCall
+	valRecv []byte
+	errRecv error
+}
+
+func (c *implSyncFetchBlobClientCall) RecvStream() interface {
+	Advance() bool
+	Value() []byte
+	Err() error
+} {
+	return implSyncFetchBlobClientCallRecv{c}
+}
+
+type implSyncFetchBlobClientCallRecv struct {
+	c *implSyncFetchBlobClientCall
+}
+
+func (c implSyncFetchBlobClientCallRecv) Advance() bool {
+	c.c.errRecv = c.c.Recv(&c.c.valRecv)
+	return c.c.errRecv == nil
+}
+func (c implSyncFetchBlobClientCallRecv) Value() []byte {
+	return c.c.valRecv
+}
+func (c implSyncFetchBlobClientCallRecv) Err() error {
+	if c.c.errRecv == io.EOF {
+		return nil
+	}
+	return c.c.errRecv
+}
+func (c *implSyncFetchBlobClientCall) Finish() (err error) {
+	err = c.ClientCall.Finish()
+	return
+}
+
+// SyncFetchBlobRecipeClientStream is the client stream for Sync.FetchBlobRecipe.
+type SyncFetchBlobRecipeClientStream interface {
+	// RecvStream returns the receiver side of the Sync.FetchBlobRecipe client stream.
+	RecvStream() interface {
+		// Advance stages an item so that it may be retrieved via Value.  Returns
+		// true iff there is an item to retrieve.  Advance must be called before
+		// Value is called.  May block if an item is not available.
+		Advance() bool
+		// Value returns the item that was staged by Advance.  May panic if Advance
+		// returned false or was not called.  Never blocks.
+		Value() ChunkHash
+		// Err returns any error encountered by Advance.  Never blocks.
+		Err() error
+	}
+}
+
+// SyncFetchBlobRecipeClientCall represents the call returned from Sync.FetchBlobRecipe.
+type SyncFetchBlobRecipeClientCall interface {
+	SyncFetchBlobRecipeClientStream
+	// Finish blocks until the server is done, and returns the positional return
+	// values for call.
+	//
+	// Finish returns immediately if the call has been canceled; depending on the
+	// timing the output could either be an error signaling cancelation, or the
+	// valid positional return values from the server.
+	//
+	// Calling Finish is mandatory for releasing stream resources, unless the call
+	// has been canceled or any of the other methods return an error.  Finish should
+	// be called at most once.
+	Finish() error
+}
+
+type implSyncFetchBlobRecipeClientCall struct {
+	rpc.ClientCall
+	valRecv ChunkHash
+	errRecv error
+}
+
+func (c *implSyncFetchBlobRecipeClientCall) RecvStream() interface {
+	Advance() bool
+	Value() ChunkHash
+	Err() error
+} {
+	return implSyncFetchBlobRecipeClientCallRecv{c}
+}
+
+type implSyncFetchBlobRecipeClientCallRecv struct {
+	c *implSyncFetchBlobRecipeClientCall
+}
+
+func (c implSyncFetchBlobRecipeClientCallRecv) Advance() bool {
+	c.c.valRecv = ChunkHash{}
+	c.c.errRecv = c.c.Recv(&c.c.valRecv)
+	return c.c.errRecv == nil
+}
+func (c implSyncFetchBlobRecipeClientCallRecv) Value() ChunkHash {
+	return c.c.valRecv
+}
+func (c implSyncFetchBlobRecipeClientCallRecv) Err() error {
+	if c.c.errRecv == io.EOF {
+		return nil
+	}
+	return c.c.errRecv
+}
+func (c *implSyncFetchBlobRecipeClientCall) Finish() (err error) {
+	err = c.ClientCall.Finish()
+	return
+}
+
+// SyncFetchChunksClientStream is the client stream for Sync.FetchChunks.
+type SyncFetchChunksClientStream interface {
+	// RecvStream returns the receiver side of the Sync.FetchChunks client stream.
+	RecvStream() interface {
+		// Advance stages an item so that it may be retrieved via Value.  Returns
+		// true iff there is an item to retrieve.  Advance must be called before
+		// Value is called.  May block if an item is not available.
+		Advance() bool
+		// Value returns the item that was staged by Advance.  May panic if Advance
+		// returned false or was not called.  Never blocks.
+		Value() ChunkData
+		// Err returns any error encountered by Advance.  Never blocks.
+		Err() error
+	}
+	// SendStream returns the send side of the Sync.FetchChunks client stream.
+	SendStream() interface {
+		// Send places the item onto the output stream.  Returns errors
+		// encountered while sending, or if Send is called after Close or
+		// the stream has been canceled.  Blocks if there is no buffer
+		// space; will unblock when buffer space is available or after
+		// the stream has been canceled.
+		Send(item ChunkHash) error
+		// Close indicates to the server that no more items will be sent;
+		// server Recv calls will receive io.EOF after all sent items.
+		// This is an optional call - e.g. a client might call Close if it
+		// needs to continue receiving items from the server after it's
+		// done sending.  Returns errors encountered while closing, or if
+		// Close is called after the stream has been canceled.  Like Send,
+		// blocks if there is no buffer space available.
+		Close() error
+	}
+}
+
+// SyncFetchChunksClientCall represents the call returned from Sync.FetchChunks.
+type SyncFetchChunksClientCall interface {
+	SyncFetchChunksClientStream
+	// Finish performs the equivalent of SendStream().Close, then blocks until
+	// the server is done, and returns the positional return values for the call.
+	//
+	// Finish returns immediately if the call has been canceled; depending on the
+	// timing the output could either be an error signaling cancelation, or the
+	// valid positional return values from the server.
+	//
+	// Calling Finish is mandatory for releasing stream resources, unless the call
+	// has been canceled or any of the other methods return an error.  Finish should
+	// be called at most once.
+	Finish() error
+}
+
+type implSyncFetchChunksClientCall struct {
+	rpc.ClientCall
+	valRecv ChunkData
+	errRecv error
+}
+
+func (c *implSyncFetchChunksClientCall) RecvStream() interface {
+	Advance() bool
+	Value() ChunkData
+	Err() error
+} {
+	return implSyncFetchChunksClientCallRecv{c}
+}
+
+type implSyncFetchChunksClientCallRecv struct {
+	c *implSyncFetchChunksClientCall
+}
+
+func (c implSyncFetchChunksClientCallRecv) Advance() bool {
+	c.c.valRecv = ChunkData{}
+	c.c.errRecv = c.c.Recv(&c.c.valRecv)
+	return c.c.errRecv == nil
+}
+func (c implSyncFetchChunksClientCallRecv) Value() ChunkData {
+	return c.c.valRecv
+}
+func (c implSyncFetchChunksClientCallRecv) Err() error {
+	if c.c.errRecv == io.EOF {
+		return nil
+	}
+	return c.c.errRecv
+}
+func (c *implSyncFetchChunksClientCall) SendStream() interface {
+	Send(item ChunkHash) error
+	Close() error
+} {
+	return implSyncFetchChunksClientCallSend{c}
+}
+
+type implSyncFetchChunksClientCallSend struct {
+	c *implSyncFetchChunksClientCall
+}
+
+func (c implSyncFetchChunksClientCallSend) Send(item ChunkHash) error {
+	return c.c.Send(item)
+}
+func (c implSyncFetchChunksClientCallSend) Close() error {
+	return c.c.CloseSend()
+}
+func (c *implSyncFetchChunksClientCall) Finish() (err error) {
+	err = c.ClientCall.Finish()
+	return
+}
+
 // SyncServerMethods is the interface a server writer
 // implements for Sync.
 //
@@ -216,9 +496,22 @@
 	// allowed to join the named SyncGroup, and if so, adds the requestor to
 	// the SyncGroup.
 	JoinSyncGroupAtAdmin(ctx *context.T, call rpc.ServerCall, sgName string, joinerName string, myInfo nosql.SyncGroupMemberInfo) (SyncGroup, error)
-	// BlobSync methods.
-	// FetchBlob returns the requested blob.
-	FetchBlob(ctx *context.T, call rpc.ServerCall, br nosql.BlobRef) error
+	// HaveBlob verifies that the peer has the requested blob, and if
+	// present, returns its size.
+	HaveBlob(ctx *context.T, call rpc.ServerCall, br nosql.BlobRef) (int64, error)
+	// FetchBlob fetches the requested blob.
+	FetchBlob(ctx *context.T, call SyncFetchBlobServerCall, br nosql.BlobRef) error
+	// Methods for incremental blob transfer. The transfer starts with the
+	// receiver making a FetchBlobRecipe call to the sender for a given
+	// BlobRef. The sender, in turn, sends the chunk hashes of all the
+	// chunks that make up the requested blob (blob recipe). The receiver
+	// looks up the chunk hashes in its local blob store, and identifies the
+	// missing ones. The receiver then fetches the missing chunks using a
+	// FetchChunks call from the sender. Finally, the receiver finishes the
+	// blob fetch by combining the chunks obtained over the network with the
+	// already available local chunks as per the blob recipe.
+	FetchBlobRecipe(ctx *context.T, call SyncFetchBlobRecipeServerCall, br nosql.BlobRef) error
+	FetchChunks(*context.T, SyncFetchChunksServerCall) error
 }
 
 // SyncServerStubMethods is the server interface containing
@@ -245,9 +538,22 @@
 	// allowed to join the named SyncGroup, and if so, adds the requestor to
 	// the SyncGroup.
 	JoinSyncGroupAtAdmin(ctx *context.T, call rpc.ServerCall, sgName string, joinerName string, myInfo nosql.SyncGroupMemberInfo) (SyncGroup, error)
-	// BlobSync methods.
-	// FetchBlob returns the requested blob.
-	FetchBlob(ctx *context.T, call rpc.ServerCall, br nosql.BlobRef) error
+	// HaveBlob verifies that the peer has the requested blob, and if
+	// present, returns its size.
+	HaveBlob(ctx *context.T, call rpc.ServerCall, br nosql.BlobRef) (int64, error)
+	// FetchBlob fetches the requested blob.
+	FetchBlob(ctx *context.T, call *SyncFetchBlobServerCallStub, br nosql.BlobRef) error
+	// Methods for incremental blob transfer. The transfer starts with the
+	// receiver making a FetchBlobRecipe call to the sender for a given
+	// BlobRef. The sender, in turn, sends the chunk hashes of all the
+	// chunks that make up the requested blob (blob recipe). The receiver
+	// looks up the chunk hashes in its local blob store, and identifies the
+	// missing ones. The receiver then fetches the missing chunks using a
+	// FetchChunks call from the sender. Finally, the receiver finishes the
+	// blob fetch by combining the chunks obtained over the network with the
+	// already available local chunks as per the blob recipe.
+	FetchBlobRecipe(ctx *context.T, call *SyncFetchBlobRecipeServerCallStub, br nosql.BlobRef) error
+	FetchChunks(*context.T, *SyncFetchChunksServerCallStub) error
 }
 
 // SyncServerStub adds universal methods to SyncServerStubMethods.
@@ -291,10 +597,22 @@
 	return s.impl.JoinSyncGroupAtAdmin(ctx, call, i0, i1, i2)
 }
 
-func (s implSyncServerStub) FetchBlob(ctx *context.T, call rpc.ServerCall, i0 nosql.BlobRef) error {
+func (s implSyncServerStub) HaveBlob(ctx *context.T, call rpc.ServerCall, i0 nosql.BlobRef) (int64, error) {
+	return s.impl.HaveBlob(ctx, call, i0)
+}
+
+func (s implSyncServerStub) FetchBlob(ctx *context.T, call *SyncFetchBlobServerCallStub, i0 nosql.BlobRef) error {
 	return s.impl.FetchBlob(ctx, call, i0)
 }
 
+func (s implSyncServerStub) FetchBlobRecipe(ctx *context.T, call *SyncFetchBlobRecipeServerCallStub, i0 nosql.BlobRef) error {
+	return s.impl.FetchBlobRecipe(ctx, call, i0)
+}
+
+func (s implSyncServerStub) FetchChunks(ctx *context.T, call *SyncFetchChunksServerCallStub) error {
+	return s.impl.FetchChunks(ctx, call)
+}
+
 func (s implSyncServerStub) Globber() *rpc.GlobState {
 	return s.gs
 }
@@ -342,12 +660,31 @@
 			Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Read"))},
 		},
 		{
-			Name: "FetchBlob",
-			Doc:  "// BlobSync methods.\n// FetchBlob returns the requested blob.",
+			Name: "HaveBlob",
+			Doc:  "// HaveBlob verifies that the peer has the requested blob, and if\n// present, returns its size.",
 			InArgs: []rpc.ArgDesc{
 				{"br", ``}, // nosql.BlobRef
 			},
-			Tags: []*vdl.Value{vdl.ValueOf(access.Tag("Read"))},
+			OutArgs: []rpc.ArgDesc{
+				{"", ``}, // int64
+			},
+		},
+		{
+			Name: "FetchBlob",
+			Doc:  "// FetchBlob fetches the requested blob.",
+			InArgs: []rpc.ArgDesc{
+				{"br", ``}, // nosql.BlobRef
+			},
+		},
+		{
+			Name: "FetchBlobRecipe",
+			Doc:  "// Methods for incremental blob transfer. The transfer starts with the\n// receiver making a FetchBlobRecipe call to the sender for a given\n// BlobRef. The sender, in turn, sends the chunk hashes of all the\n// chunks that make up the requested blob (blob recipe). The receiver\n// looks up the chunk hashes in its local blob store, and identifies the\n// missing ones. The receiver then fetches the missing chunks using a\n// FetchChunks call from the sender. Finally, the receiver finishes the\n// blob fetch by combining the chunks obtained over the network with the\n// already available local chunks as per the blob recipe.",
+			InArgs: []rpc.ArgDesc{
+				{"br", ``}, // nosql.BlobRef
+			},
+		},
+		{
+			Name: "FetchChunks",
 		},
 	},
 }
@@ -436,3 +773,174 @@
 func (s implSyncGetDeltasServerCallSend) Send(item DeltaResp) error {
 	return s.s.Send(item)
 }
+
+// SyncFetchBlobServerStream is the server stream for Sync.FetchBlob.
+type SyncFetchBlobServerStream interface {
+	// SendStream returns the send side of the Sync.FetchBlob server stream.
+	SendStream() interface {
+		// Send places the item onto the output stream.  Returns errors encountered
+		// while sending.  Blocks if there is no buffer space; will unblock when
+		// buffer space is available.
+		Send(item []byte) error
+	}
+}
+
+// SyncFetchBlobServerCall represents the context passed to Sync.FetchBlob.
+type SyncFetchBlobServerCall interface {
+	rpc.ServerCall
+	SyncFetchBlobServerStream
+}
+
+// SyncFetchBlobServerCallStub is a wrapper that converts rpc.StreamServerCall into
+// a typesafe stub that implements SyncFetchBlobServerCall.
+type SyncFetchBlobServerCallStub struct {
+	rpc.StreamServerCall
+}
+
+// Init initializes SyncFetchBlobServerCallStub from rpc.StreamServerCall.
+func (s *SyncFetchBlobServerCallStub) Init(call rpc.StreamServerCall) {
+	s.StreamServerCall = call
+}
+
+// SendStream returns the send side of the Sync.FetchBlob server stream.
+func (s *SyncFetchBlobServerCallStub) SendStream() interface {
+	Send(item []byte) error
+} {
+	return implSyncFetchBlobServerCallSend{s}
+}
+
+type implSyncFetchBlobServerCallSend struct {
+	s *SyncFetchBlobServerCallStub
+}
+
+func (s implSyncFetchBlobServerCallSend) Send(item []byte) error {
+	return s.s.Send(item)
+}
+
+// SyncFetchBlobRecipeServerStream is the server stream for Sync.FetchBlobRecipe.
+type SyncFetchBlobRecipeServerStream interface {
+	// SendStream returns the send side of the Sync.FetchBlobRecipe server stream.
+	SendStream() interface {
+		// Send places the item onto the output stream.  Returns errors encountered
+		// while sending.  Blocks if there is no buffer space; will unblock when
+		// buffer space is available.
+		Send(item ChunkHash) error
+	}
+}
+
+// SyncFetchBlobRecipeServerCall represents the context passed to Sync.FetchBlobRecipe.
+type SyncFetchBlobRecipeServerCall interface {
+	rpc.ServerCall
+	SyncFetchBlobRecipeServerStream
+}
+
+// SyncFetchBlobRecipeServerCallStub is a wrapper that converts rpc.StreamServerCall into
+// a typesafe stub that implements SyncFetchBlobRecipeServerCall.
+type SyncFetchBlobRecipeServerCallStub struct {
+	rpc.StreamServerCall
+}
+
+// Init initializes SyncFetchBlobRecipeServerCallStub from rpc.StreamServerCall.
+func (s *SyncFetchBlobRecipeServerCallStub) Init(call rpc.StreamServerCall) {
+	s.StreamServerCall = call
+}
+
+// SendStream returns the send side of the Sync.FetchBlobRecipe server stream.
+func (s *SyncFetchBlobRecipeServerCallStub) SendStream() interface {
+	Send(item ChunkHash) error
+} {
+	return implSyncFetchBlobRecipeServerCallSend{s}
+}
+
+type implSyncFetchBlobRecipeServerCallSend struct {
+	s *SyncFetchBlobRecipeServerCallStub
+}
+
+func (s implSyncFetchBlobRecipeServerCallSend) Send(item ChunkHash) error {
+	return s.s.Send(item)
+}
+
+// SyncFetchChunksServerStream is the server stream for Sync.FetchChunks.
+type SyncFetchChunksServerStream interface {
+	// RecvStream returns the receiver side of the Sync.FetchChunks server stream.
+	RecvStream() interface {
+		// Advance stages an item so that it may be retrieved via Value.  Returns
+		// true iff there is an item to retrieve.  Advance must be called before
+		// Value is called.  May block if an item is not available.
+		Advance() bool
+		// Value returns the item that was staged by Advance.  May panic if Advance
+		// returned false or was not called.  Never blocks.
+		Value() ChunkHash
+		// Err returns any error encountered by Advance.  Never blocks.
+		Err() error
+	}
+	// SendStream returns the send side of the Sync.FetchChunks server stream.
+	SendStream() interface {
+		// Send places the item onto the output stream.  Returns errors encountered
+		// while sending.  Blocks if there is no buffer space; will unblock when
+		// buffer space is available.
+		Send(item ChunkData) error
+	}
+}
+
+// SyncFetchChunksServerCall represents the context passed to Sync.FetchChunks.
+type SyncFetchChunksServerCall interface {
+	rpc.ServerCall
+	SyncFetchChunksServerStream
+}
+
+// SyncFetchChunksServerCallStub is a wrapper that converts rpc.StreamServerCall into
+// a typesafe stub that implements SyncFetchChunksServerCall.
+type SyncFetchChunksServerCallStub struct {
+	rpc.StreamServerCall
+	valRecv ChunkHash
+	errRecv error
+}
+
+// Init initializes SyncFetchChunksServerCallStub from rpc.StreamServerCall.
+func (s *SyncFetchChunksServerCallStub) Init(call rpc.StreamServerCall) {
+	s.StreamServerCall = call
+}
+
+// RecvStream returns the receiver side of the Sync.FetchChunks server stream.
+func (s *SyncFetchChunksServerCallStub) RecvStream() interface {
+	Advance() bool
+	Value() ChunkHash
+	Err() error
+} {
+	return implSyncFetchChunksServerCallRecv{s}
+}
+
+type implSyncFetchChunksServerCallRecv struct {
+	s *SyncFetchChunksServerCallStub
+}
+
+func (s implSyncFetchChunksServerCallRecv) Advance() bool {
+	s.s.valRecv = ChunkHash{}
+	s.s.errRecv = s.s.Recv(&s.s.valRecv)
+	return s.s.errRecv == nil
+}
+func (s implSyncFetchChunksServerCallRecv) Value() ChunkHash {
+	return s.s.valRecv
+}
+func (s implSyncFetchChunksServerCallRecv) Err() error {
+	if s.s.errRecv == io.EOF {
+		return nil
+	}
+	return s.s.errRecv
+}
+
+// SendStream returns the send side of the Sync.FetchChunks server stream.
+func (s *SyncFetchChunksServerCallStub) SendStream() interface {
+	Send(item ChunkData) error
+} {
+	return implSyncFetchChunksServerCallSend{s}
+}
+
+type implSyncFetchChunksServerCallSend struct {
+	s *SyncFetchChunksServerCallStub
+}
+
+func (s implSyncFetchChunksServerCallSend) Send(item ChunkData) error {
+	return s.s.Send(item)
+}
diff --git a/x/ref/services/syncbase/server/interfaces/sync_types.vdl b/x/ref/services/syncbase/server/interfaces/sync_types.vdl
index 6d608a3..324928b 100644
--- a/x/ref/services/syncbase/server/interfaces/sync_types.vdl
+++ b/x/ref/services/syncbase/server/interfaces/sync_types.vdl
@@ -117,3 +117,13 @@
 	Rec     LogRec
 	RespVec GenVector
 }
+
+// ChunkHash contains the hash of a chunk that is part of a blob's recipe.
+type ChunkHash struct {
+	Hash []byte
+}
+
+// ChunkData contains the data of a chunk.
+type ChunkData struct {
+	Data []byte
+}
diff --git a/x/ref/services/syncbase/server/interfaces/sync_types.vdl.go b/x/ref/services/syncbase/server/interfaces/sync_types.vdl.go
index aeb4e31..8ef80a8 100644
--- a/x/ref/services/syncbase/server/interfaces/sync_types.vdl.go
+++ b/x/ref/services/syncbase/server/interfaces/sync_types.vdl.go
@@ -228,6 +228,26 @@
 func (x DeltaRespRespVec) Name() string                    { return "RespVec" }
 func (x DeltaRespRespVec) __VDLReflect(__DeltaRespReflect) {}
 
+// ChunkHash contains the hash of a chunk that is part of a blob's recipe.
+type ChunkHash struct {
+	Hash []byte
+}
+
+func (ChunkHash) __VDLReflect(struct {
+	Name string `vdl:"v.io/syncbase/x/ref/services/syncbase/server/interfaces.ChunkHash"`
+}) {
+}
+
+// ChunkData contains the data of a chunk.
+type ChunkData struct {
+	Data []byte
+}
+
+func (ChunkData) __VDLReflect(struct {
+	Name string `vdl:"v.io/syncbase/x/ref/services/syncbase/server/interfaces.ChunkData"`
+}) {
+}
+
 func init() {
 	vdl.Register((*PrefixGenVector)(nil))
 	vdl.Register((*GenVector)(nil))
@@ -238,6 +258,8 @@
 	vdl.Register((*SyncGroup)(nil))
 	vdl.Register((*DeltaReq)(nil))
 	vdl.Register((*DeltaResp)(nil))
+	vdl.Register((*ChunkHash)(nil))
+	vdl.Register((*ChunkData)(nil))
 }
 
 const NoGroupId = GroupId(0)
diff --git a/x/ref/services/syncbase/server/nosql/database.go b/x/ref/services/syncbase/server/nosql/database.go
index 1fe3066..94b37d8 100644
--- a/x/ref/services/syncbase/server/nosql/database.go
+++ b/x/ref/services/syncbase/server/nosql/database.go
@@ -15,8 +15,6 @@
 	wire "v.io/syncbase/v23/services/syncbase/nosql"
 	"v.io/syncbase/v23/syncbase/nosql/query_db"
 	"v.io/syncbase/v23/syncbase/nosql/query_exec"
-	"v.io/syncbase/x/ref/services/syncbase/localblobstore"
-	"v.io/syncbase/x/ref/services/syncbase/localblobstore/fs_cablobstore"
 	"v.io/syncbase/x/ref/services/syncbase/server/interfaces"
 	"v.io/syncbase/x/ref/services/syncbase/server/util"
 	"v.io/syncbase/x/ref/services/syncbase/server/watchable"
@@ -45,9 +43,6 @@
 	// and do not actually open the store in NewDatabase.
 	st store.Store // stores all data for a single database
 
-	// Local blob store associated with this database.
-	bst localblobstore.BlobStore
-
 	// Active snapshots and transactions corresponding to client batches.
 	// TODO(sadovsky): Add timeouts and GC.
 	mu  sync.Mutex // protects the fields below
@@ -93,17 +88,11 @@
 	if err != nil {
 		return nil, err
 	}
-	// Open a co-located blob store, adjacent to the structured store.
-	bst, err := fs_cablobstore.Create(ctx, path.Join(opts.RootDir, "blobs"))
-	if err != nil {
-		return nil, err
-	}
 	return &database{
 		name:   name,
 		a:      a,
 		exists: true,
 		st:     st,
-		bst:    bst,
 		sns:    make(map[uint64]store.Snapshot),
 		txs:    make(map[uint64]store.Transaction),
 	}, nil
@@ -379,13 +368,6 @@
 	return d.st
 }
 
-func (d *database) BlobSt() localblobstore.BlobStore {
-	if !d.exists {
-		vlog.Fatalf("database %q does not exist", d.name)
-	}
-	return d.bst
-}
-
 func (d *database) App() interfaces.App {
 	return d.a
 }
diff --git a/x/ref/services/syncbase/server/service.go b/x/ref/services/syncbase/server/service.go
index bafa388..180692f 100644
--- a/x/ref/services/syncbase/server/service.go
+++ b/x/ref/services/syncbase/server/service.go
@@ -130,7 +130,7 @@
 	}
 	// Note, vsync.New internally handles both first-time and subsequent
 	// invocations.
-	if s.sync, err = vsync.New(ctx, call, s, opts.Server); err != nil {
+	if s.sync, err = vsync.New(ctx, call, s, opts.Server, opts.RootDir); err != nil {
 		return nil, err
 	}
 	return s, nil
diff --git a/x/ref/services/syncbase/vsync/blob.go b/x/ref/services/syncbase/vsync/blob.go
index 56a4eed..ff04066 100644
--- a/x/ref/services/syncbase/vsync/blob.go
+++ b/x/ref/services/syncbase/vsync/blob.go
@@ -8,8 +8,11 @@
 	"io"
 
 	wire "v.io/syncbase/v23/services/syncbase/nosql"
-	"v.io/syncbase/x/ref/services/syncbase/localblobstore"
+	blob "v.io/syncbase/x/ref/services/syncbase/localblobstore"
+	"v.io/syncbase/x/ref/services/syncbase/server/interfaces"
+	"v.io/syncbase/x/ref/services/syncbase/server/util"
 	"v.io/v23/context"
+	"v.io/v23/naming"
 	"v.io/v23/rpc"
 	"v.io/v23/verror"
 	"v.io/x/lib/vlog"
@@ -19,11 +22,25 @@
 	chunkSize = 8 * 1024
 )
 
+// blobLocInfo contains the location information about a BlobRef. This location
+// information is merely a hint used to search for the blob.
+type blobLocInfo struct {
+	peer   string                          // Syncbase from which the presence of this BlobRef was first learned.
+	source string                          // Syncbase that originated this blob.
+	sgIds  map[interfaces.GroupId]struct{} // SyncGroups through which the BlobRef was learned.
+}
+
 ////////////////////////////////////////////////////////////
 // RPCs for managing blobs between Syncbase and its clients.
 
 func (sd *syncDatabase) CreateBlob(ctx *context.T, call rpc.ServerCall) (wire.BlobRef, error) {
-	bst := sd.db.BlobSt()
+	vlog.VI(2).Infof("sync: CreateBlob: begin")
+	defer vlog.VI(2).Infof("sync: CreateBlob: end")
+
+	// Get this Syncbase's blob store handle.
+	ss := sd.sync.(*syncService)
+	bst := ss.bst
+
 	writer, err := bst.NewBlobWriter(ctx, "")
 	if err != nil {
 		return wire.NullBlobRef, err
@@ -31,13 +48,19 @@
 	defer writer.CloseWithoutFinalize()
 
 	name := writer.Name()
-	vlog.VI(2).Infof("sync: CreateBlob: blob ref %s", name)
+	vlog.VI(4).Infof("sync: CreateBlob: blob ref %s", name)
 	return wire.BlobRef(name), nil
 }
 
 func (sd *syncDatabase) PutBlob(ctx *context.T, call wire.BlobManagerPutBlobServerCall, br wire.BlobRef) error {
-	bst := sd.db.BlobSt()
-	writer, err := bst.NewBlobWriter(ctx, string(br))
+	vlog.VI(2).Infof("sync: PutBlob: begin br %v", br)
+	defer vlog.VI(2).Infof("sync: PutBlob: end br %v", br)
+
+	// Get this Syncbase's blob store handle.
+	ss := sd.sync.(*syncService)
+	bst := ss.bst
+
+	writer, err := bst.ResumeBlobWriter(ctx, string(br))
 	if err != nil {
 		return err
 	}
@@ -45,7 +68,7 @@
 
 	stream := call.RecvStream()
 	for stream.Advance() {
-		item := localblobstore.BlockOrFile{Block: stream.Value()}
+		item := blob.BlockOrFile{Block: stream.Value()}
 		if err = writer.AppendFragment(item); err != nil {
 			return err
 		}
@@ -54,8 +77,14 @@
 }
 
 func (sd *syncDatabase) CommitBlob(ctx *context.T, call rpc.ServerCall, br wire.BlobRef) error {
-	bst := sd.db.BlobSt()
-	writer, err := bst.NewBlobWriter(ctx, string(br))
+	vlog.VI(2).Infof("sync: CommitBlob: begin br %v", br)
+	defer vlog.VI(2).Infof("sync: CommitBlob: end br %v", br)
+
+	// Get this Syncbase's blob store handle.
+	ss := sd.sync.(*syncService)
+	bst := ss.bst
+
+	writer, err := bst.ResumeBlobWriter(ctx, string(br))
 	if err != nil {
 		return err
 	}
@@ -63,7 +92,13 @@
 }
 
 func (sd *syncDatabase) GetBlobSize(ctx *context.T, call rpc.ServerCall, br wire.BlobRef) (int64, error) {
-	bst := sd.db.BlobSt()
+	vlog.VI(2).Infof("sync: GetBlobSize: begin br %v", br)
+	defer vlog.VI(2).Infof("sync: GetBlobSize: end br %v", br)
+
+	// Get this Syncbase's blob store handle.
+	ss := sd.sync.(*syncService)
+	bst := ss.bst
+
 	reader, err := bst.NewBlobReader(ctx, string(br))
 	if err != nil {
 		return 0, err
@@ -78,47 +113,46 @@
 }
 
 func (sd *syncDatabase) GetBlob(ctx *context.T, call wire.BlobManagerGetBlobServerCall, br wire.BlobRef, offset int64) error {
+	vlog.VI(2).Infof("sync: GetBlob: begin br %v", br)
+	defer vlog.VI(2).Infof("sync: GetBlob: end br %v", br)
+
 	// First get the blob locally if available.
-	err := sd.getLocalBlob(ctx, call, br, offset)
-	if err == nil {
-		return nil
-	}
-
-	// Locate the blob on a remote blob store and fetch it.
-	return verror.NewErrNotImplemented(ctx)
-}
-
-// getLocalBlob looks for a blob in the local store and, if found, reads the
-// blob and sends it to the client.  If the blob is found, it starts reading it
-// from the given offset and sends its bytes into the client stream.
-func (sd *syncDatabase) getLocalBlob(ctx *context.T, call wire.BlobManagerGetBlobServerCall, br wire.BlobRef, offset int64) error {
-	bst := sd.db.BlobSt()
-	reader, err := bst.NewBlobReader(ctx, string(br))
-	if err != nil {
+	ss := sd.sync.(*syncService)
+	err := getLocalBlob(ctx, call.SendStream(), ss.bst, br, offset)
+	if err == nil || verror.ErrorID(err) == wire.ErrBlobNotCommitted.ID {
 		return err
 	}
-	defer reader.Close()
 
-	buf := make([]byte, chunkSize)
-	stream := call.SendStream()
-
-	for {
-		nbytes, err := reader.ReadAt(buf, offset)
-		if err != nil && err != io.EOF {
-			return err
-		}
-		if err == io.EOF || nbytes <= 0 {
-			break
-		}
-		offset += int64(nbytes)
-		stream.Send(buf[:nbytes])
-	}
-
-	return nil
+	return sd.fetchBlobRemote(ctx, br, nil, call, offset)
 }
 
 func (sd *syncDatabase) FetchBlob(ctx *context.T, call wire.BlobManagerFetchBlobServerCall, br wire.BlobRef, priority uint64) error {
-	return verror.NewErrNotImplemented(ctx)
+	vlog.VI(2).Infof("sync: FetchBlob: begin br %v", br)
+	defer vlog.VI(2).Infof("sync: FetchBlob: end br %v", br)
+
+	clientStream := call.SendStream()
+
+	// Check if BlobRef already exists locally.
+	ss := sd.sync.(*syncService)
+	bst := ss.bst
+
+	bReader, err := bst.NewBlobReader(ctx, string(br))
+	if err == nil {
+		finalized := bReader.IsFinalized()
+		bReader.Close()
+
+		if !finalized {
+			return wire.NewErrBlobNotCommitted(ctx)
+		}
+		clientStream.Send(wire.BlobFetchStatus{State: wire.BlobFetchStateDone})
+		return nil
+	}
+
+	// Wait for this blob's turn.
+	// TODO(hpucha): Implement a blob queue.
+	clientStream.Send(wire.BlobFetchStatus{State: wire.BlobFetchStatePending})
+
+	return sd.fetchBlobRemote(ctx, br, call, nil, 0)
 }
 
 func (sd *syncDatabase) PinBlob(ctx *context.T, call rpc.ServerCall, br wire.BlobRef) error {
@@ -136,6 +170,260 @@
 ////////////////////////////////////////////////////////////
 // RPC for blob fetch between Syncbases.
 
-func (s *syncService) FetchBlob(ctx *context.T, call rpc.ServerCall, br wire.BlobRef) error {
+func (s *syncService) FetchBlob(ctx *context.T, call interfaces.SyncFetchBlobServerCall, br wire.BlobRef) error {
+	vlog.VI(2).Infof("sync: FetchBlob: sb-sb begin br %v", br)
+	defer vlog.VI(2).Infof("sync: FetchBlob: sb-sb end br %v", br)
+	return getLocalBlob(ctx, call.SendStream(), s.bst, br, 0)
+}
+
+func (s *syncService) HaveBlob(ctx *context.T, call rpc.ServerCall, br wire.BlobRef) (int64, error) {
+	vlog.VI(2).Infof("sync: HaveBlob: begin br %v", br)
+	defer vlog.VI(2).Infof("sync: HaveBlob: end br %v", br)
+
+	bReader, err := s.bst.NewBlobReader(ctx, string(br))
+	if err != nil {
+		return 0, err
+	}
+	defer bReader.Close()
+	if !bReader.IsFinalized() {
+		return 0, wire.NewErrBlobNotCommitted(ctx)
+	}
+	return bReader.Size(), nil
+}
+
+func (s *syncService) FetchBlobRecipe(ctx *context.T, call interfaces.SyncFetchBlobRecipeServerCall, br wire.BlobRef) error {
 	return verror.NewErrNotImplemented(ctx)
 }
+
+func (s *syncService) FetchChunks(ctx *context.T, call interfaces.SyncFetchChunksServerCall) error {
+	return verror.NewErrNotImplemented(ctx)
+}
+
+////////////////////////////////////////////////////////////
+// Helpers.
+
+type byteStream interface {
+	Send(item []byte) error
+}
+
+// getLocalBlob looks for a blob in the local store and, if found, reads the
+// blob and sends it to the client.  If the blob is found, it starts reading it
+// from the given offset and sends its bytes into the client stream.
+func getLocalBlob(ctx *context.T, stream byteStream, bst blob.BlobStore, br wire.BlobRef, offset int64) error {
+	vlog.VI(4).Infof("sync: getLocalBlob: begin br %v, offset %v", br, offset)
+	defer vlog.VI(4).Infof("sync: getLocalBlob: end br %v, offset %v", br, offset)
+
+	reader, err := bst.NewBlobReader(ctx, string(br))
+	if err != nil {
+		return err
+	}
+	defer reader.Close()
+
+	if !reader.IsFinalized() {
+		return wire.NewErrBlobNotCommitted(ctx)
+	}
+
+	buf := make([]byte, chunkSize)
+	for {
+		nbytes, err := reader.ReadAt(buf, offset)
+		if err != nil && err != io.EOF {
+			return err
+		}
+		if nbytes <= 0 {
+			break
+		}
+		offset += int64(nbytes)
+		stream.Send(buf[:nbytes])
+		if err == io.EOF {
+			break
+		}
+	}
+
+	return nil
+}
+
+func (sd *syncDatabase) fetchBlobRemote(ctx *context.T, br wire.BlobRef, statusCall wire.BlobManagerFetchBlobServerCall, dataCall wire.BlobManagerGetBlobServerCall, offset int64) error {
+	vlog.VI(4).Infof("sync: fetchBlobRemote: begin br %v, offset %v", br, offset)
+	defer vlog.VI(4).Infof("sync: fetchBlobRemote: end br %v, offset %v", br, offset)
+
+	var sendStatus, sendData bool
+	var statusStream interface {
+		Send(item wire.BlobFetchStatus) error
+	}
+	var dataStream interface {
+		Send(item []byte) error
+	}
+
+	if statusCall != nil {
+		sendStatus = true
+		statusStream = statusCall.SendStream()
+	}
+	if dataCall != nil {
+		sendData = true
+		dataStream = dataCall.SendStream()
+	}
+
+	if sendStatus {
+		// Start blob source discovery.
+		statusStream.Send(wire.BlobFetchStatus{State: wire.BlobFetchStateLocating})
+	}
+
+	// Locate blob.
+	peer, size, err := sd.locateBlob(ctx, br)
+	if err != nil {
+		return err
+	}
+
+	// Start blob fetching.
+	status := wire.BlobFetchStatus{State: wire.BlobFetchStateFetching, Total: size}
+	if sendStatus {
+		statusStream.Send(status)
+	}
+
+	ss := sd.sync.(*syncService)
+	bst := ss.bst
+
+	bWriter, err := bst.NewBlobWriter(ctx, string(br))
+	if err != nil {
+		return err
+	}
+
+	c := interfaces.SyncClient(peer)
+	ctxPeer, cancel := context.WithRootCancel(ctx)
+	stream, err := c.FetchBlob(ctxPeer, br)
+	if err == nil {
+		peerStream := stream.RecvStream()
+		for peerStream.Advance() {
+			item := blob.BlockOrFile{Block: peerStream.Value()}
+			if err = bWriter.AppendFragment(item); err != nil {
+				break
+			}
+			curSize := int64(len(item.Block))
+			status.Received += curSize
+			if sendStatus {
+				statusStream.Send(status)
+			}
+			if sendData {
+				if curSize <= offset {
+					offset -= curSize
+				} else if offset != 0 {
+					dataStream.Send(item.Block[offset:])
+					offset = 0
+				} else {
+					dataStream.Send(item.Block)
+				}
+			}
+		}
+
+		if err != nil {
+			cancel()
+			stream.Finish()
+		} else {
+			err = peerStream.Err()
+			if terr := stream.Finish(); err == nil {
+				err = terr
+			}
+			cancel()
+		}
+	}
+
+	bWriter.Close()
+	if err != nil {
+		// Clean up the blob with failed download, so that it can be
+		// downloaded again. Ignore any error from deletion.
+		bst.DeleteBlob(ctx, string(br))
+	} else {
+		status := wire.BlobFetchStatus{State: wire.BlobFetchStateDone}
+		if sendStatus {
+			statusStream.Send(status)
+		}
+	}
+	return err
+}
+
+// TODO(hpucha): Add syncgroup driven blob discovery.
+func (sd *syncDatabase) locateBlob(ctx *context.T, br wire.BlobRef) (string, int64, error) {
+	vlog.VI(4).Infof("sync: locateBlob: begin br %v", br)
+	defer vlog.VI(4).Infof("sync: locateBlob: end br %v", br)
+
+	ss := sd.sync.(*syncService)
+	loc, err := ss.getBlobLocInfo(ctx, br)
+	if err != nil {
+		return "", 0, err
+	}
+
+	// Search for blob amongst the source peer and peer learned from.
+	var peers = []string{loc.source, loc.peer}
+	for _, p := range peers {
+		vlog.VI(4).Infof("sync: locateBlob: attempting %s", p)
+		// Get the mounttables for this peer.
+		mtTables, err := sd.getMountTables(ctx, p)
+		if err != nil {
+			continue
+		}
+
+		for mt := range mtTables {
+			absName := naming.Join(mt, p, util.SyncbaseSuffix)
+			c := interfaces.SyncClient(absName)
+			size, err := c.HaveBlob(ctx, br)
+			if err == nil {
+				vlog.VI(4).Infof("sync: locateBlob: found blob on %s", absName)
+				return absName, size, nil
+			}
+		}
+	}
+
+	return "", 0, verror.New(verror.ErrInternal, ctx, "blob not found")
+
+}
+
+func (sd *syncDatabase) getMountTables(ctx *context.T, peer string) (map[string]struct{}, error) {
+	ss := sd.sync.(*syncService)
+	mInfo := ss.copyMemberInfo(ctx, peer)
+
+	mtTables := make(map[string]struct{})
+	for gdbName, sgInfo := range mInfo.db2sg {
+		appName, dbName, err := splitAppDbName(ctx, gdbName)
+		if err != nil {
+			return nil, err
+		}
+		st, err := ss.getDbStore(ctx, nil, appName, dbName)
+		if err != nil {
+			return nil, err
+		}
+
+		for id := range sgInfo {
+			sg, err := getSyncGroupById(ctx, st, id)
+			if err != nil {
+				continue
+			}
+			if _, ok := sg.Joiners[peer]; !ok {
+				// Peer is no longer part of the SyncGroup.
+				continue
+			}
+			for _, mt := range sg.Spec.MountTables {
+				mtTables[mt] = struct{}{}
+			}
+		}
+	}
+	return mtTables, nil
+}
+
+// TODO(hpucha): Persist the blob directory periodically.
+func (s *syncService) addBlobLocInfo(ctx *context.T, br wire.BlobRef, info *blobLocInfo) error {
+	s.blobDirLock.Lock()
+	defer s.blobDirLock.Unlock()
+
+	s.blobDirectory[br] = info
+	return nil
+}
+
+func (s *syncService) getBlobLocInfo(ctx *context.T, br wire.BlobRef) (*blobLocInfo, error) {
+	s.blobDirLock.Lock()
+	defer s.blobDirLock.Unlock()
+
+	if info, ok := s.blobDirectory[br]; ok {
+		return info, nil
+	}
+	return nil, verror.New(verror.ErrInternal, ctx, "blob state not found", br)
+}
diff --git a/x/ref/services/syncbase/vsync/initiator.go b/x/ref/services/syncbase/vsync/initiator.go
index 5d3d9b8..cccf217 100644
--- a/x/ref/services/syncbase/vsync/initiator.go
+++ b/x/ref/services/syncbase/vsync/initiator.go
@@ -15,14 +15,16 @@
 	"strings"
 	"time"
 
+	"v.io/syncbase/v23/services/syncbase/nosql"
 	"v.io/syncbase/x/ref/services/syncbase/server/interfaces"
 	"v.io/syncbase/x/ref/services/syncbase/server/util"
 	"v.io/syncbase/x/ref/services/syncbase/server/watchable"
 	"v.io/syncbase/x/ref/services/syncbase/store"
 	"v.io/v23/context"
 	"v.io/v23/naming"
+	"v.io/v23/vdl"
 	"v.io/v23/verror"
-	"v.io/x/lib/set"
+	"v.io/v23/vom"
 	"v.io/x/lib/vlog"
 )
 
@@ -62,10 +64,6 @@
 func (s *syncService) syncer(ctx *context.T) {
 	defer s.pending.Done()
 
-	// TODO(hpucha): Do we need context per initiator round?
-	ctx, cancel := context.WithRootCancel(ctx)
-	defer cancel()
-
 	ticker := time.NewTicker(peerSyncInterval)
 	defer ticker.Stop()
 
@@ -106,11 +104,13 @@
 // initiation round), the work done by the initiator is idempotent.
 //
 // TODO(hpucha): Check the idempotence, esp in addNode in DAG.
-func (s *syncService) getDeltasFromPeer(ctx *context.T, peer string) {
+func (s *syncService) getDeltasFromPeer(ctxIn *context.T, peer string) {
 	vlog.VI(2).Infof("sync: getDeltasFromPeer: begin: contacting peer %s", peer)
 	defer vlog.VI(2).Infof("sync: getDeltasFromPeer: end: contacting peer %s", peer)
 
-	info := s.allMembers.members[peer]
+	ctx, cancel := context.WithRootCancel(ctxIn)
+
+	info := s.copyMemberInfo(ctx, peer)
 	if info == nil {
 		vlog.Fatalf("sync: getDeltasFromPeer: missing information in member view for %q", peer)
 	}
@@ -167,6 +167,8 @@
 			// Returning here since something could be wrong with
 			// the connection, and no point in attempting the next
 			// Database.
+			cancel()
+			stream.Finish()
 			return
 		}
 		vlog.VI(3).Infof("sync: getDeltasFromPeer: got reply: %v", iSt.remote)
@@ -182,8 +184,11 @@
 	if connected {
 		stream.Finish()
 	}
+	cancel()
 }
 
+type sgSet map[interfaces.GroupId]struct{}
+
 // initiationState is accumulated for each Database during an initiation round.
 type initiationState struct {
 	// Relative name of the peer to sync with.
@@ -193,10 +198,11 @@
 	mtTables map[string]struct{}
 
 	// SyncGroups being requested in the initiation round.
-	sgIds map[interfaces.GroupId]struct{}
+	sgIds sgSet
 
-	// SyncGroup prefixes being requested in the initiation round.
-	sgPfxs map[string]struct{}
+	// SyncGroup prefixes being requested in the initiation round, and their
+	// corresponding SyncGroup ids.
+	sgPfxs map[string]sgSet
 
 	// Local generation vector.
 	local interfaces.GenVector
@@ -267,8 +273,8 @@
 // SyncGroups in the specified Database.
 func (iSt *initiationState) peerMtTblsAndSgInfo(ctx *context.T, peer string, info sgMemberInfo) {
 	iSt.mtTables = make(map[string]struct{})
-	iSt.sgIds = make(map[interfaces.GroupId]struct{})
-	iSt.sgPfxs = make(map[string]struct{})
+	iSt.sgIds = make(sgSet)
+	iSt.sgPfxs = make(map[string]sgSet)
 
 	for id := range info {
 		sg, err := getSyncGroupById(ctx, iSt.st, id)
@@ -285,7 +291,12 @@
 		iSt.sgIds[id] = struct{}{}
 
 		for _, p := range sg.Spec.Prefixes {
-			iSt.sgPfxs[p] = struct{}{}
+			sgs, ok := iSt.sgPfxs[p]
+			if !ok {
+				sgs = make(sgSet)
+				iSt.sgPfxs[p] = sgs
+			}
+			sgs[id] = struct{}{}
 		}
 	}
 }
@@ -340,7 +351,12 @@
 	}
 	localPfxs := extractAndSortPrefixes(local)
 
-	sgPfxs := set.String.ToSlice(iSt.sgPfxs)
+	sgPfxs := make([]string, len(iSt.sgPfxs))
+	i := 0
+	for p := range iSt.sgPfxs {
+		sgPfxs[i] = p
+		i++
+	}
 	sort.Strings(sgPfxs)
 
 	iSt.local = make(interfaces.GenVector)
@@ -454,6 +470,12 @@
 			if err := iSt.insertRecInLogDagAndDb(ctx, rec, batchId, v.Value.Value, tx); err != nil {
 				return err
 			}
+
+			// Check for BlobRefs, and process them.
+			if err := iSt.processBlobRefs(ctx, &rec.Metadata, v.Value.Value); err != nil {
+				return err
+			}
+
 			// Mark object dirty.
 			iSt.updObjects[rec.Metadata.ObjId] = &objConflictState{}
 		}
@@ -494,6 +516,68 @@
 	return err
 }
 
+func (iSt *initiationState) processBlobRefs(ctx *context.T, m *interfaces.LogRecMetadata, valbuf []byte) error {
+	objid := m.ObjId
+	srcPeer := syncbaseIdToName(m.Id)
+
+	vlog.VI(4).Infof("sync: processBlobRefs: begin processing blob refs for objid %s", objid)
+	defer vlog.VI(4).Infof("sync: processBlobRefs: end processing blob refs for objid %s", objid)
+
+	if valbuf == nil {
+		return nil
+	}
+
+	var val *vdl.Value
+	if err := vom.Decode(valbuf, &val); err != nil {
+		return err
+	}
+
+	brs := make(map[nosql.BlobRef]struct{})
+	if err := extractBlobRefs(val, brs); err != nil {
+		return err
+	}
+	sgIds := make(sgSet)
+	for br := range brs {
+		for p, sgs := range iSt.sgPfxs {
+			if strings.HasPrefix(extractAppKey(objid), p) {
+				for sg := range sgs {
+					sgIds[sg] = struct{}{}
+				}
+			}
+		}
+		vlog.VI(4).Infof("sync: processBlobRefs: Found blobref %v peer %v, source %v, sgs %v", br, iSt.peer, srcPeer, sgIds)
+		info := &blobLocInfo{peer: iSt.peer, source: srcPeer, sgIds: sgIds}
+		if err := iSt.sync.addBlobLocInfo(ctx, br, info); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// TODO(hpucha): Handle blobrefs part of list, map, any.
+func extractBlobRefs(val *vdl.Value, brs map[nosql.BlobRef]struct{}) error {
+	if val == nil {
+		return nil
+	}
+	switch val.Kind() {
+	case vdl.String:
+		// Could be a BlobRef.
+		var br nosql.BlobRef
+		if val.Type() == vdl.TypeOf(br) {
+			brs[nosql.BlobRef(val.RawString())] = struct{}{}
+		}
+	case vdl.Struct:
+		for i := 0; i < val.Type().NumField(); i++ {
+			v := val.StructField(i)
+			if err := extractBlobRefs(v, brs); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 // insertRecInLogDagAndDb adds a new log record to log and dag data structures,
 // and inserts the versioned value in the Database.
 func (iSt *initiationState) insertRecInLogDagAndDb(ctx *context.T, rec *localLogRec, batchId uint64, valbuf []byte, tx store.Transaction) error {
diff --git a/x/ref/services/syncbase/vsync/initiator_test.go b/x/ref/services/syncbase/vsync/initiator_test.go
index c3f2538..af09ce1 100644
--- a/x/ref/services/syncbase/vsync/initiator_test.go
+++ b/x/ref/services/syncbase/vsync/initiator_test.go
@@ -22,9 +22,65 @@
 	"v.io/syncbase/x/ref/services/syncbase/server/interfaces"
 	"v.io/syncbase/x/ref/services/syncbase/server/util"
 	"v.io/syncbase/x/ref/services/syncbase/server/watchable"
+	"v.io/v23/vdl"
+	"v.io/v23/vom"
 	_ "v.io/x/ref/runtime/factories/generic"
 )
 
+func TestExtractBlobRefs(t *testing.T) {
+	var tests [][]byte
+	br := nosql.BlobRef("123")
+
+	// BlobRef is the value.
+	buf0, err := vom.Encode(br)
+	if err != nil {
+		t.Fatalf("Encode(BlobRef) failed, err %v", err)
+	}
+	tests = append(tests, buf0)
+
+	// Struct contains BlobRef.
+	type test1Struct struct {
+		A int64
+		B string
+		C nosql.BlobRef
+	}
+	v1 := test1Struct{A: 10, B: "foo", C: br}
+	buf1, err := vom.Encode(v1)
+	if err != nil {
+		t.Fatalf("Encode(test1Struct) failed, err %v", err)
+	}
+	tests = append(tests, buf1)
+
+	// Nested struct contains BlobRef.
+	type test2Struct struct {
+		A int64
+		B string
+		C test1Struct
+	}
+	v2 := test2Struct{A: 10, B: "foo", C: v1}
+	buf2, err := vom.Encode(v2)
+	if err != nil {
+		t.Fatalf("Encode(test2Struct) failed, err %v", err)
+	}
+	tests = append(tests, buf2)
+
+	for i, buf := range tests {
+		var val *vdl.Value
+		if err := vom.Decode(buf, &val); err != nil {
+			t.Fatalf("Decode failed (test %d), err %v", i, err)
+		}
+
+		gotbrs := make(map[nosql.BlobRef]struct{})
+		if err := extractBlobRefs(val, gotbrs); err != nil {
+			t.Fatalf("extractBlobRefs failed (test %d), err %v", i, err)
+		}
+		wantbrs := map[nosql.BlobRef]struct{}{br: struct{}{}}
+		if !reflect.DeepEqual(gotbrs, wantbrs) {
+			t.Fatalf("Data mismatch in blobrefs (test %d), got %v, want %v", i, gotbrs, wantbrs)
+		}
+	}
+}
+
 // TestLogStreamRemoteOnly tests processing of a remote log stream. Commands are
 // in file testdata/remote-init-00.log.sync.
 func TestLogStreamRemoteOnly(t *testing.T) {
@@ -99,8 +155,12 @@
 
 	// Verify Database state.
 	valbuf, err := svc.St().Get([]byte(objid), nil)
-	if err != nil || string(valbuf) != "abc" {
-		t.Fatalf("Invalid object %s in Database %v, err %v", objid, string(valbuf), err)
+	var val string
+	if err := vom.Decode(valbuf, &val); err != nil {
+		t.Fatalf("Value decode failed, err %v", err)
+	}
+	if err != nil || val != "abc" {
+		t.Fatalf("Invalid object %s in Database %v, err %v", objid, val, err)
 	}
 	tx := svc.St().NewTransaction()
 	version, err := watchable.GetVersion(nil, tx, []byte(objid))
@@ -192,8 +252,12 @@
 
 	// Verify Database state.
 	valbuf, err := svc.St().Get([]byte(objid), nil)
-	if err != nil || string(valbuf) != "abc" {
-		t.Fatalf("Invalid object %s in Database %v, err %v", objid, string(valbuf), err)
+	var val string
+	if err := vom.Decode(valbuf, &val); err != nil {
+		t.Fatalf("Value decode failed, err %v", err)
+	}
+	if err != nil || val != "abc" {
+		t.Fatalf("Invalid object %s in Database %v, err %v", objid, val, err)
 	}
 	tx := svc.St().NewTransaction()
 	versbuf, err := watchable.GetVersion(nil, tx, []byte(objid))
@@ -234,7 +298,11 @@
 
 	// Verify Database state.
 	valbuf, err := svc.St().Get([]byte(objid), nil)
-	if err != nil || string(valbuf) != "abc" {
+	var val string
+	if err := vom.Decode(valbuf, &val); err != nil {
+		t.Fatalf("Value decode failed, err %v", err)
+	}
+	if err != nil || val != "abc" {
 		t.Fatalf("Invalid object %s in Database %v, err %v", objid, string(valbuf), err)
 	}
 	tx := svc.St().NewTransaction()
@@ -277,7 +345,11 @@
 
 	// Verify Database state.
 	valbuf, err := svc.St().Get([]byte(objid), nil)
-	if err != nil || string(valbuf) != "abc" {
+	var val string
+	if err := vom.Decode(valbuf, &val); err != nil {
+		t.Fatalf("Value decode failed, err %v", err)
+	}
+	if err != nil || val != "abc" {
 		t.Fatalf("Invalid object %s in Database %v, err %v", objid, string(valbuf), err)
 	}
 	tx := svc.St().NewTransaction()
@@ -345,7 +417,7 @@
 		t.Fatalf("newInitiationState failed with err %v", err)
 	}
 
-	testIfMapArrEqual(t, iSt.sgPfxs, sg1.Spec.Prefixes)
+	testIfSgPfxsEqual(t, iSt.sgPfxs, sg1.Spec.Prefixes)
 	testIfMapArrEqual(t, iSt.mtTables, sg1.Spec.MountTables)
 
 	s.initDbSyncStateInMem(nil, "mockapp", "mockdb")
@@ -375,14 +447,31 @@
 	return svc, iSt, cleanup
 }
 
-func testIfMapArrEqual(t *testing.T, m map[string]struct{}, a []string) {
-	if len(a) != len(m) {
-		t.Fatalf("testIfMapArrEqual diff lengths, got %v want %v", a, m)
+func testIfSgPfxsEqual(t *testing.T, m map[string]sgSet, a []string) {
+	aMap := arrToMap(a)
+
+	if len(aMap) != len(m) {
+		t.Fatalf("testIfSgPfxsEqual diff lengths, got %v want %v", aMap, m)
 	}
 
-	for _, p := range a {
+	for p := range aMap {
 		if _, ok := m[p]; !ok {
-			t.Fatalf("testIfMapArrEqual want %v", p)
+			t.Fatalf("testIfSgPfxsEqual want %v", p)
 		}
 	}
 }
+
+func testIfMapArrEqual(t *testing.T, m map[string]struct{}, a []string) {
+	aMap := arrToMap(a)
+	if !reflect.DeepEqual(m, aMap) {
+		t.Fatalf("testIfMapArrEqual failed map %v, arr %v", m, aMap)
+	}
+}
+
+func arrToMap(a []string) map[string]struct{} {
+	m := make(map[string]struct{})
+	for _, s := range a {
+		m[s] = struct{}{}
+	}
+	return m
+}
diff --git a/x/ref/services/syncbase/vsync/replay_test.go b/x/ref/services/syncbase/vsync/replay_test.go
index 98b7d20..03d6dc6 100644
--- a/x/ref/services/syncbase/vsync/replay_test.go
+++ b/x/ref/services/syncbase/vsync/replay_test.go
@@ -22,6 +22,7 @@
 	"v.io/syncbase/x/ref/services/syncbase/server/util"
 	"v.io/syncbase/x/ref/services/syncbase/server/watchable"
 	"v.io/v23/context"
+	"v.io/v23/vom"
 )
 
 const (
@@ -359,9 +360,15 @@
 			t.Fatalf("createReplayStream unknown command %v", cmd)
 		}
 
+		var val string = "abc"
+		valbuf, err := vom.Encode(val)
+		if err != nil {
+			t.Fatalf("createReplayStream encode failed, err %v", err)
+		}
+
 		rec := interfaces.DeltaRespRec{interfaces.LogRec{
 			Metadata: createMetadata(t, ty, cmd),
-			Value:    []byte("abc"),
+			Value:    valbuf,
 		}}
 
 		stream.add(rec)
diff --git a/x/ref/services/syncbase/vsync/responder.go b/x/ref/services/syncbase/vsync/responder.go
index 80c2c52..c417eca 100644
--- a/x/ref/services/syncbase/vsync/responder.go
+++ b/x/ref/services/syncbase/vsync/responder.go
@@ -11,7 +11,6 @@
 
 	wire "v.io/syncbase/v23/services/syncbase/nosql"
 	"v.io/syncbase/x/ref/services/syncbase/server/interfaces"
-	"v.io/syncbase/x/ref/services/syncbase/server/util"
 	"v.io/syncbase/x/ref/services/syncbase/server/watchable"
 	"v.io/syncbase/x/ref/services/syncbase/store"
 	"v.io/v23/context"
@@ -446,11 +445,7 @@
 	// The key starts with one of the store's reserved prefixes for managed
 	// namespaces (e.g. $row, $perms).  Remove that prefix before comparing
 	// it with the SyncGroup prefixes which are defined by the application.
-	parts := util.SplitKeyParts(rec.Metadata.ObjId)
-	if len(parts) < 2 {
-		vlog.Fatalf("sync: filterLogRec: invalid entry key %s", rec.Metadata.ObjId)
-	}
-	key := util.JoinKeyParts(parts[1:]...)
+	key := extractAppKey(rec.Metadata.ObjId)
 
 	filter := true
 	var maxGen uint64
diff --git a/x/ref/services/syncbase/vsync/sync.go b/x/ref/services/syncbase/vsync/sync.go
index ebb55f6..fd65c66 100644
--- a/x/ref/services/syncbase/vsync/sync.go
+++ b/x/ref/services/syncbase/vsync/sync.go
@@ -14,9 +14,13 @@
 import (
 	"fmt"
 	"math/rand"
+	"path"
 	"sync"
 	"time"
 
+	"v.io/syncbase/v23/services/syncbase/nosql"
+	blob "v.io/syncbase/x/ref/services/syncbase/localblobstore"
+	fsblob "v.io/syncbase/x/ref/services/syncbase/localblobstore/fs_cablobstore"
 	"v.io/syncbase/x/ref/services/syncbase/server/interfaces"
 	"v.io/syncbase/x/ref/services/syncbase/server/util"
 	"v.io/syncbase/x/ref/services/syncbase/store"
@@ -62,7 +66,8 @@
 	// names must be advertised in the appropriate mount tables.
 
 	// In-memory sync membership info aggregated across databases.
-	allMembers *memberView
+	allMembers     *memberView
+	allMembersLock sync.RWMutex
 
 	// In-memory sync state per Database. This state is populated at
 	// startup, and periodically persisted by the initiator.
@@ -75,13 +80,20 @@
 	// access to the batch set.
 	batchesLock sync.Mutex
 	batches     batchSet
+
+	// Metadata related to blob handling.
+	bst           blob.BlobStore                 // local blob store associated with this Syncbase.
+	blobDirectory map[nosql.BlobRef]*blobLocInfo // directory structure containing blob location information.
+	blobDirLock   sync.RWMutex                   // lock to synchronize access to the blob directory information.
+
 }
 
 // syncDatabase contains the metadata for syncing a database. This struct is
 // used as a receiver to hand off the app-initiated SyncGroup calls that arrive
 // against a nosql.Database to the sync module.
 type syncDatabase struct {
-	db interfaces.Database
+	db   interfaces.Database
+	sync interfaces.SyncServerMethods
 }
 
 var (
@@ -111,7 +123,7 @@
 // changes to its objects. The "initiator" thread is responsible for
 // periodically contacting peers to fetch changes from them. In addition, the
 // sync module responds to incoming RPCs from remote sync modules.
-func New(ctx *context.T, call rpc.ServerCall, sv interfaces.Service, server rpc.Server) (*syncService, error) {
+func New(ctx *context.T, call rpc.ServerCall, sv interfaces.Service, server rpc.Server, rootDir string) (*syncService, error) {
 	s := &syncService{
 		sv:      sv,
 		server:  server,
@@ -136,13 +148,21 @@
 
 	// data.Id is now guaranteed to be initialized.
 	s.id = data.Id
-	s.name = fmt.Sprintf("%x", s.id)
+	s.name = syncbaseIdToName(s.id)
 
 	// Initialize in-memory state for the sync module before starting any threads.
 	if err := s.initSync(ctx); err != nil {
 		return nil, verror.New(verror.ErrInternal, ctx, err)
 	}
 
+	// Open a blob store.
+	var err error
+	s.bst, err = fsblob.Create(ctx, path.Join(rootDir, "blobs"))
+	if err != nil {
+		return nil, err
+	}
+	s.blobDirectory = make(map[nosql.BlobRef]*blobLocInfo)
+
 	// Channel to propagate close event to all threads.
 	s.closed = make(chan struct{})
 	s.pending.Add(2)
@@ -159,12 +179,17 @@
 // Close cleans up sync state.
 // TODO(hpucha): Hook it up to server shutdown of syncbased.
 func (s *syncService) Close() {
+	s.bst.Close()
 	close(s.closed)
 	s.pending.Wait()
 }
 
+func syncbaseIdToName(id uint64) string {
+	return fmt.Sprintf("%x", id)
+}
+
 func NewSyncDatabase(db interfaces.Database) *syncDatabase {
-	return &syncDatabase{db: db}
+	return &syncDatabase{db: db, sync: db.App().Service().Sync()}
 }
 
 func (s *syncService) stKey() string {
diff --git a/x/ref/services/syncbase/vsync/syncgroup.go b/x/ref/services/syncbase/vsync/syncgroup.go
index bf4e3aa..905a319 100644
--- a/x/ref/services/syncbase/vsync/syncgroup.go
+++ b/x/ref/services/syncbase/vsync/syncgroup.go
@@ -263,6 +263,9 @@
 // getMembers returns all SyncGroup members and the count of SyncGroups each one
 // joined.
 func (s *syncService) getMembers(ctx *context.T) map[string]uint32 {
+	s.allMembersLock.Lock()
+	defer s.allMembersLock.Unlock()
+
 	s.refreshMembersIfExpired(ctx)
 
 	members := make(map[string]uint32)
@@ -277,6 +280,28 @@
 	return members
 }
 
+// copyMemberInfo returns a copy of the info for the requested peer.
+func (s *syncService) copyMemberInfo(ctx *context.T, member string) *memberInfo {
+	s.allMembersLock.RLock()
+	defer s.allMembersLock.RUnlock()
+
+	info, ok := s.allMembers.members[member]
+	if !ok {
+		return nil
+	}
+
+	// Make a copy.
+	infoCopy := &memberInfo{make(map[string]sgMemberInfo)}
+	for gdbName, sgInfo := range info.db2sg {
+		infoCopy.db2sg[gdbName] = make(sgMemberInfo)
+		for gid, mi := range sgInfo {
+			infoCopy.db2sg[gdbName][gid] = mi
+		}
+	}
+
+	return infoCopy
+}
+
 // Low-level utility functions to access DB entries without tracking their
 // relationships.
 // Use the functions above to manipulate SyncGroups.
@@ -395,7 +420,7 @@
 		// has Admin privilege.
 
 		// Get this Syncbase's sync module handle.
-		ss := sd.db.App().Service().Sync().(*syncService)
+		ss := sd.sync.(*syncService)
 
 		// Instantiate sg. Add self as joiner.
 		sg := &interfaces.SyncGroup{
@@ -485,7 +510,7 @@
 	// Brand new join.
 
 	// Get this Syncbase's sync module handle.
-	ss := sd.db.App().Service().Sync().(*syncService)
+	ss := sd.sync.(*syncService)
 
 	// Contact a SyncGroup Admin to join the SyncGroup.
 	sg = &interfaces.SyncGroup{}
@@ -739,7 +764,7 @@
 
 func (sd *syncDatabase) publishInMountTables(ctx *context.T, call rpc.ServerCall, spec wire.SyncGroupSpec) error {
 	// Get this Syncbase's sync module handle.
-	ss := sd.db.App().Service().Sync().(*syncService)
+	ss := sd.sync.(*syncService)
 
 	for _, mt := range spec.MountTables {
 		name := naming.Join(mt, ss.name)
diff --git a/x/ref/services/syncbase/vsync/test_util.go b/x/ref/services/syncbase/vsync/test_util.go
index b06b785..a323f19 100644
--- a/x/ref/services/syncbase/vsync/test_util.go
+++ b/x/ref/services/syncbase/vsync/test_util.go
@@ -14,8 +14,6 @@
 	"time"
 
 	wire "v.io/syncbase/v23/services/syncbase/nosql"
-	"v.io/syncbase/x/ref/services/syncbase/localblobstore"
-	"v.io/syncbase/x/ref/services/syncbase/localblobstore/fs_cablobstore"
 	"v.io/syncbase/x/ref/services/syncbase/server/interfaces"
 	"v.io/syncbase/x/ref/services/syncbase/server/util"
 	"v.io/syncbase/x/ref/services/syncbase/server/watchable"
@@ -33,7 +31,6 @@
 	engine   string
 	dir      string
 	st       store.Store
-	bst      localblobstore.BlobStore
 	sync     *syncService
 	shutdown func()
 }
@@ -47,7 +44,7 @@
 }
 
 func (s *mockService) App(ctx *context.T, call rpc.ServerCall, appName string) (interfaces.App, error) {
-	return &mockApp{st: s.st, bst: s.bst}, nil
+	return &mockApp{st: s.st}, nil
 }
 
 func (s *mockService) AppNames(ctx *context.T, call rpc.ServerCall) ([]string, error) {
@@ -56,12 +53,11 @@
 
 // mockApp emulates a Syncbase App.  It is used to access a mock database.
 type mockApp struct {
-	st  store.Store
-	bst localblobstore.BlobStore
+	st store.Store
 }
 
 func (a *mockApp) NoSQLDatabase(ctx *context.T, call rpc.ServerCall, dbName string) (interfaces.Database, error) {
-	return &mockDatabase{st: a.st, bst: a.bst}, nil
+	return &mockDatabase{st: a.st}, nil
 }
 
 func (a *mockApp) NoSQLDatabaseNames(ctx *context.T, call rpc.ServerCall) ([]string, error) {
@@ -90,18 +86,13 @@
 
 // mockDatabase emulates a Syncbase Database.  It is used to test sync functionality.
 type mockDatabase struct {
-	st  store.Store
-	bst localblobstore.BlobStore
+	st store.Store
 }
 
 func (d *mockDatabase) St() store.Store {
 	return d.st
 }
 
-func (d *mockDatabase) BlobSt() localblobstore.BlobStore {
-	return d.bst
-}
-
 func (d *mockDatabase) CheckPermsInternal(ctx *context.T, call rpc.ServerCall, st store.StoreReader) error {
 	return verror.NewErrNotImplemented(ctx)
 }
@@ -132,20 +123,14 @@
 	st, err = watchable.Wrap(st, &watchable.Options{
 		ManagedPrefixes: []string{util.RowPrefix, util.PermsPrefix},
 	})
-	bst, err := fs_cablobstore.Create(ctx, path.Join(dir, "blobs"))
-	if err != nil {
-		t.Fatalf("cannot create blob store (%s): %v", dir, err)
-	}
 
 	s := &mockService{
 		st:       st,
-		bst:      bst,
 		engine:   engine,
 		dir:      dir,
 		shutdown: shutdown,
 	}
-	if s.sync, err = New(ctx, nil, s, nil); err != nil {
-		bst.Close()
+	if s.sync, err = New(ctx, nil, s, nil, dir); err != nil {
 		util.DestroyStore(engine, dir)
 		t.Fatalf("cannot create sync service: %v", err)
 	}
@@ -156,7 +141,6 @@
 func destroyService(t *testing.T, s *mockService) {
 	defer s.shutdown()
 	defer s.sync.Close()
-	s.bst.Close()
 	if err := util.DestroyStore(s.engine, s.dir); err != nil {
 		t.Fatalf("cannot destroy store %s (%s): %v", s.engine, s.dir, err)
 	}
diff --git a/x/ref/services/syncbase/vsync/util.go b/x/ref/services/syncbase/vsync/util.go
index af0b158..2f15cf8 100644
--- a/x/ref/services/syncbase/vsync/util.go
+++ b/x/ref/services/syncbase/vsync/util.go
@@ -9,6 +9,7 @@
 import (
 	"time"
 
+	"v.io/syncbase/x/ref/services/syncbase/server/util"
 	"v.io/syncbase/x/ref/services/syncbase/store"
 	"v.io/v23/context"
 	"v.io/v23/rpc"
@@ -81,3 +82,17 @@
 	}
 	return time.Unix(timestamp/nanoPerSec, timestamp%nanoPerSec)
 }
+
+// extractAppKey extracts the app key from the key sent over the wire between
+// two Syncbases. The on-wire key starts with one of the store's reserved
+// prefixes for managed namespaces (e.g. $row, $perms). This function removes
+// that prefix and returns the application component of the key. This is done
+// typically before comparing keys with the SyncGroup prefixes which are defined
+// by the application.
+func extractAppKey(key string) string {
+	parts := util.SplitKeyParts(key)
+	if len(parts) < 2 {
+		vlog.Fatalf("sync: extractAppKey: invalid entry key %s", key)
+	}
+	return util.JoinKeyParts(parts[1:]...)
+}