Repository: googleforgames/open-saves
Pull request #266 (2021-06-28 20:41:32)

feat: Metadata operations for chunked blobs
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://github.com/googleforgames/open-saves/blob/main/docs/contributing.md and developer guide https://github.com/googleforgames/open-saves/blob/main/docs/development.md
2. Please label this pull request according to what type of issue you are addressing.
3. Ensure you have added or ran the appropriate tests for your PR.
-->

**What type of PR is this?**
/kind feat

**What this PR does / Why we need it**:
This PR adds metadata operations necessary to implement chunked blobs.

**Which issue(s) this PR fixes**:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Closes #<issue number>`, or `Closes (paste link of issue)`.
-->
Closes #

**Special notes for your reviewer**:
The server implementation will be submitted as another PR.


Patch:
diff --git a/internal/pkg/metadb/blobref/blobref.go b/internal/pkg/metadb/blobref/blobref.go
index 96fec83..adf2dbf 100644
--- a/internal/pkg/metadb/blobref/blobref.go
+++ b/internal/pkg/metadb/blobref/blobref.go
@@ -34,6 +34,8 @@ type BlobRef struct {
 	// It can be non-existent (e.g. deleted already) but then the Status
 	// should not be Blob StatusReady.
 	RecordKey string
+	// Chunked is whether the BlobRef is chunked or not.
+	Chunked bool
 
 	// Timestamps keeps track of creation and modification times and stores a randomly
 	// generated UUID to maintain consistency.
@@ -86,6 +88,15 @@ func NewBlobRef(size int64, storeKey, recordKey string) *BlobRef {
 	}
 }
 
+// NewChunkedBlobRef creates a new BlobRef object with Size and Chunked
+// set 0 and true, respectively.
+// Other behaviors are the same as NewBlobRef
+func NewChunkedBlobRef(storeKey, recordKey string) *BlobRef {
+	b := NewBlobRef(0, storeKey, recordKey)
+	b.Chunked = true
+	return b
+}
+
 // ObjectPath returns an object path for the backend blob storage.
 func (b *BlobRef) ObjectPath() string {
 	return b.Key.String()
diff --git a/internal/pkg/metadb/blobref/chunkref/chunkref.go b/internal/pkg/metadb/blobref/chunkref/chunkref.go
index 6013abd..dd4302f 100644
--- a/internal/pkg/metadb/blobref/chunkref/chunkref.go
+++ b/internal/pkg/metadb/blobref/chunkref/chunkref.go
@@ -21,15 +21,26 @@ import (
 	"cloud.google.com/go/datastore"
 	"github.com/google/uuid"
 	"github.com/googleforgames/open-saves/internal/pkg/cache"
+	"github.com/googleforgames/open-saves/internal/pkg/metadb/blobref"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/timestamps"
 )
 
+// ChunkRef is a metadata entity to keep track of chunks stored in an external blob store.
+// It is a child entity of and always associated to a BlobRef.
 type ChunkRef struct {
-	Key     uuid.UUID `datastore:"-"`
+	// Key is the primary key of the ChunkRef.
+	Key uuid.UUID `datastore:"-"`
+	// BlobRef is the key of parent BlobRef.
 	BlobRef uuid.UUID `datastore:"-"`
 
-	Number     int32
-	Size       int32
+	// Number is the position of the chunk in the BlobRef.
+	Number int32
+	// Size is the byte size of the chunk.
+	Size int32
+	// Status is the current status of the chunk.
+	blobref.Status
+	// Timestamps keeps track of creation and modification times and stores a randomly
+	// generated UUID to maintain consistency.
 	Timestamps timestamps.Timestamps
 }
 
@@ -72,12 +83,13 @@ func (c *ChunkRef) ObjectPath() string {
 }
 
 // New creates a new ChunkRef instance with the input parameters.
-func New(blobRef uuid.UUID, number, size int32) *ChunkRef {
+func New(blobRef uuid.UUID, number int32) *ChunkRef {
 	return &ChunkRef{
-		Key:     uuid.New(),
-		BlobRef: blobRef,
-		Number:  number,
-		Size:    size,
+		Key:        uuid.New(),
+		BlobRef:    blobRef,
+		Number:     number,
+		Status:     blobref.StatusInitializing,
+		Timestamps: timestamps.New(),
 	}
 }
 
diff --git a/internal/pkg/metadb/metadb.go b/internal/pkg/metadb/metadb.go
index f5d1a94..25efcbd 100644
--- a/internal/pkg/metadb/metadb.go
+++ b/internal/pkg/metadb/metadb.go
@@ -21,6 +21,7 @@ import (
 	ds "cloud.google.com/go/datastore"
 	"github.com/google/uuid"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/blobref"
+	"github.com/googleforgames/open-saves/internal/pkg/metadb/blobref/chunkref"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/record"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/store"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/timestamps"
@@ -36,6 +37,7 @@ const (
 	storeKind  = "store"
 	recordKind = "record"
 	blobKind   = "blob"
+	chunkKind  = "chunk"
 
 	propertiesField = "Properties"
 	tagsField       = "Tags"
@@ -97,9 +99,17 @@ func (d *MetaDB) createBlobKey(key uuid.UUID) *ds.Key {
 	return k
 }
 
+func (m *MetaDB) createChunkRefKey(blobRefKey, key uuid.UUID) *ds.Key {
+	bk := m.createBlobKey(blobRefKey)
+	bk.Namespace = m.Namespace
+	ck := ds.NameKey(chunkKind, key.String(), bk)
+	ck.Namespace = m.Namespace
+	return ck
+}
+
 func (m *MetaDB) getBlobRef(ctx context.Context, tx *ds.Transaction, key uuid.UUID) (*blobref.BlobRef, error) {
 	if key == uuid.Nil {
-		return nil, status.Error(codes.FailedPrecondition, "there is no an external blob associated")
+		return nil, status.Error(codes.FailedPrecondition, "there is no external blob associated")
 	}
 	blob := new(blobref.BlobRef)
 	if tx == nil {
@@ -115,8 +125,8 @@ func (m *MetaDB) getBlobRef(ctx context.Context, tx *ds.Transaction, key uuid.UU
 }
 
 // Returns a modified Record and the caller must commit the change.
-func (m *MetaDB) markBlobRefForDeletion(_ context.Context, tx *ds.Transaction,
-	storeKey string, record *record.Record, blob *blobref.BlobRef, newBlobKey uuid.UUID) (*record.Record, error) {
+func (m *MetaDB) markBlobRefForDeletion(tx *ds.Transaction,
+	record *record.Record, blob *blobref.BlobRef, newBlobKey uuid.UUID) (*record.Record, error) {
 	if record.ExternalBlob == uuid.Nil {
 		return nil, status.Error(codes.FailedPrecondition, "the record doesn't have an external blob associated")
 	}
@@ -293,7 +303,7 @@ func (m *MetaDB) UpdateRecord(ctx context.Context, storeKey string, key string,
 			if err != nil {
 				return err
 			}
-			toUpdate, err = m.markBlobRefForDeletion(ctx, tx, storeKey, toUpdate, oldBlob, uuid.Nil)
+			toUpdate, err = m.markBlobRefForDeletion(tx, toUpdate, oldBlob, uuid.Nil)
 			if err != nil {
 				return err
 			}
@@ -335,7 +345,7 @@ func (m *MetaDB) DeleteRecord(ctx context.Context, storeKey, key string) error {
 		if record.ExternalBlob != uuid.Nil {
 			blob, err := m.getBlobRef(ctx, tx, record.ExternalBlob)
 			if err == nil {
-				_, err = m.markBlobRefForDeletion(ctx, tx, storeKey, record, blob, uuid.Nil)
+				_, err = m.markBlobRefForDeletion(tx, record, blob, uuid.Nil)
 				if err != nil {
 					return err
 				}
@@ -394,6 +404,16 @@ func (m *MetaDB) GetBlobRef(ctx context.Context, key uuid.UUID) (*blobref.BlobRe
 	return m.getBlobRef(ctx, nil, key)
 }
 
+func (m *MetaDB) getCurrentBlobRef(ctx context.Context, tx *ds.Transaction, storeKey, recordKey string) (*blobref.BlobRef, error) {
+	record := new(record.Record)
+	err := tx.Get(m.createRecordKey(storeKey, recordKey), record)
+	if err != nil {
+		return nil, err
+	}
+	blob, err := m.getBlobRef(ctx, tx, record.ExternalBlob)
+	return blob, err
+}
+
 // GetCurrentBlobRef gets a BlobRef object associated with a record.
 // Returned errors:
 // 	- NotFound: the record is not found.
@@ -401,17 +421,43 @@ func (m *MetaDB) GetBlobRef(ctx context.Context, key uuid.UUID) (*blobref.BlobRe
 func (m *MetaDB) GetCurrentBlobRef(ctx context.Context, storeKey, recordKey string) (*blobref.BlobRef, error) {
 	var blob *blobref.BlobRef
 	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
-		record := new(record.Record)
-		err := tx.Get(m.createRecordKey(storeKey, recordKey), record)
-		if err != nil {
-			return err
-		}
-		blob, err = m.getBlobRef(ctx, tx, record.ExternalBlob)
+		var err error
+		blob, err = m.getCurrentBlobRef(ctx, tx, storeKey, recordKey)
 		return err
-	})
+	}, ds.ReadOnly)
 	return blob, datastoreErrToGRPCStatus(err)
 }
 
+func (m *MetaDB) getReadyChunks(ctx context.Context, tx *ds.Transaction, blob *blobref.BlobRef) ([]*chunkref.ChunkRef, error) {
+	query := m.newQuery(chunkKind).Transaction(tx).Ancestor(m.createBlobKey(blob.Key)).
+		Filter("Status =", int(blobref.StatusReady))
+	iter := m.client.Run(ctx, query)
+	chunks := []*chunkref.ChunkRef{}
+	for {
+		chunk := new(chunkref.ChunkRef)
+		_, err := iter.Next(chunk)
+		if err == iterator.Done {
+			break
+		} else if err != nil {
+			return nil, err
+		}
+		chunks = append(chunks, chunk)
+	}
+	return chunks, nil
+}
+
+func (m *MetaDB) chunkObjectsSizeSum(ctx context.Context, tx *ds.Transaction, blob *blobref.BlobRef) (int64, error) {
+	chunks, err := m.getReadyChunks(ctx, tx, blob)
+	if err != nil {
+		return 0, err
+	}
+	size := int64(0)
+	for _, chunk := range chunks {
+		size += int64(chunk.Size)
+	}
+	return size, nil
+}
+
 // PromoteBlobRefToCurrent promotes the provided BlobRef object as a current
 // external blob reference.
 // Returned errors:
@@ -434,26 +480,37 @@ func (m *MetaDB) PromoteBlobRefToCurrent(ctx context.Context, blob *blobref.Blob
 			if err != nil {
 				return err
 			}
-			record, err = m.markBlobRefForDeletion(ctx, tx, blob.StoreKey, record, oldBlob, blob.Key)
+			record, err = m.markBlobRefForDeletion(tx, record, oldBlob, blob.Key)
 			if err != nil {
 				return err
 			}
 		}
-		record.BlobSize = blob.Size
-		record.ExternalBlob = blob.Key
-		if err := m.mutateSingleInTransaction(tx, ds.NewUpdate(rkey, record)); err != nil {
-			return err
-		}
 
+		// Update the blob size for chunked uploads
+		if blob.Chunked {
+			// TODO(yuryu): should check if chunks are continuous?
+			size, err := m.chunkObjectsSizeSum(ctx, tx, blob)
+			if err != nil {
+				return err
+			}
+			blob.Size = size
+		}
 		if blob.Status != blobref.StatusReady {
 			if blob.Ready() != nil {
 				return status.Error(codes.Internal, "blob is not ready to become current")
 			}
-			blob.Timestamps.Update()
-			if _, err := tx.Mutate(ds.NewUpdate(m.createBlobKey(blob.Key), blob)); err != nil {
-				return err
-			}
 		}
+		blob.Timestamps.Update()
+		if _, err := tx.Mutate(ds.NewUpdate(m.createBlobKey(blob.Key), blob)); err != nil {
+			return err
+		}
+
+		record.BlobSize = blob.Size
+		record.ExternalBlob = blob.Key
+		if err := m.mutateSingleInTransaction(tx, ds.NewUpdate(rkey, record)); err != nil {
+			return err
+		}
+
 		return nil
 	})
 	if err != nil {
@@ -492,7 +549,27 @@ func (m *MetaDB) RemoveBlobFromRecord(ctx context.Context, storeKey string, reco
 		if err != nil {
 			return err
 		}
-		record, err = m.markBlobRefForDeletion(ctx, tx, storeKey, record, blob, uuid.Nil)
+
+		if blob.Chunked {
+			// Mark child chunks as well
+			chunks, err := m.getReadyChunks(ctx, tx, blob)
+			if err != nil {
+				return err
+			}
+			muts := []*ds.Mutation{}
+			for _, chunk := range chunks {
+				if err := chunk.MarkForDeletion(); err != nil {
+					return err
+				}
+				chunk.Timestamps.Update()
+				muts = append(muts, ds.NewUpdate(m.createChunkRefKey(chunk.BlobRef, chunk.Key), chunk))
+			}
+			if _, err := tx.Mutate(muts...); err != nil {
+				return err
+			}
+		}
+
+		record, err = m.markBlobRefForDeletion(tx, record, blob, uuid.Nil)
 		if err != nil {
 			return err
 		}
@@ -504,6 +581,24 @@ func (m *MetaDB) RemoveBlobFromRecord(ctx context.Context, storeKey string, reco
 	return record, blob, nil
 }
 
+func (m *MetaDB) deleteChildChunkRefs(ctx context.Context, tx *ds.Transaction, blob *blobref.BlobRef) error {
+	query := m.newQuery(chunkKind).Ancestor(m.createBlobKey(blob.Key)).Transaction(tx).KeysOnly()
+	iter := m.client.Run(ctx, query)
+	for {
+		key, err := iter.Next(nil)
+		if err == iterator.Done {
+			break
+		}
+		if err != nil {
+			return err
+		}
+		if err := tx.Delete(key); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // DeleteBlobRef deletes the BlobRef object from the database.
 // Returned errors:
 //	- NotFound: the blobref object is not found
@@ -517,6 +612,11 @@ func (m *MetaDB) DeleteBlobRef(ctx context.Context, key uuid.UUID) error {
 		if blob.Status == blobref.StatusReady {
 			return status.Error(codes.FailedPrecondition, "blob is currently marked as ready. mark it for deletion first")
 		}
+		if blob.Chunked {
+			if err := m.deleteChildChunkRefs(ctx, tx, blob); err != nil {
+				return err
+			}
+		}
 		return tx.Delete(m.createBlobKey(key))
 	})
 	return datastoreErrToGRPCStatus(err)
@@ -580,3 +680,128 @@ func (m *MetaDB) QueryRecords(ctx context.Context, filters []*pb.QueryFilter, st
 	}
 	return match, keys, nil
 }
+
+func (m *MetaDB) findChunkRefsByNumber(ctx context.Context, tx *ds.Transaction, storeKey, recordKey string, blobKey uuid.UUID, number int32) ([]*chunkref.ChunkRef, error) {
+	query := m.newQuery(chunkKind).Transaction(tx).Ancestor(m.createBlobKey(blobKey)).Filter("Number = ", number)
+	iter := m.client.Run(ctx, query)
+
+	chunks := []*chunkref.ChunkRef{}
+	for {
+		chunk := new(chunkref.ChunkRef)
+		_, err := iter.Next(chunk)
+		if err == iterator.Done {
+			break
+		} else if err != nil {
+			return nil, err
+		}
+		chunks = append(chunks, chunk)
+	}
+	return chunks, nil
+}
+
+// FindChunkRefByNumber returns a ChunkRef object for the specified store, record, and number.
+// The ChunkRef must be Ready and the chunk upload session must be committed.
+func (m *MetaDB) FindChunkRefByNumber(ctx context.Context, storeKey, recordKey string, number int32) (*chunkref.ChunkRef, error) {
+	chunks := []*chunkref.ChunkRef{}
+	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
+		blob, err := m.getCurrentBlobRef(ctx, tx, storeKey, recordKey)
+		if err != nil {
+			return err
+		}
+		chunks, err = m.findChunkRefsByNumber(ctx, tx, storeKey, recordKey, blob.Key, number)
+		return err
+	}, ds.ReadOnly)
+	if err != nil {
+		return nil, datastoreErrToGRPCStatus(err)
+	}
+	for _, c := range chunks {
+		if c.Status == blobref.StatusReady {
+			return c, nil
+		}
+	}
+	return nil, status.Errorf(codes.NotFound, "chunk number (%v) was not found for record (%v)", number, recordKey)
+}
+
+// MarkChunkRefReady marks the specified chunk as Ready. If the current session has another chunk
+// with the same Number, it will be marked for deletion.
+func (m *MetaDB) MarkChunkRefReady(ctx context.Context, chunk *chunkref.ChunkRef) error {
+	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
+		blob, err := m.getBlobRef(ctx, tx, chunk.BlobRef)
+		if err != nil {
+			return err
+		}
+
+		// Mark the chunk ready (Datastore's transaction isolation guarantees this is not visible until
+		// we commit the transaction).
+		if err := chunk.Ready(); err != nil {
+			return err
+		}
+		chunk.Timestamps.Update()
+		if err := m.mutateSingleInTransaction(tx, ds.NewUpdate(m.createChunkRefKey(blob.Key, chunk.Key), chunk)); err != nil {
+			return err
+		}
+
+		// Mark any other (should be at most one though) Ready chunks for deletion.
+		otherChunks, err := m.findChunkRefsByNumber(ctx, tx, blob.StoreKey, blob.RecordKey, chunk.BlobRef, chunk.Number)
+		if err != nil {
+			return err
+		}
+		for _, o := range otherChunks {
+			if o.Status == blobref.StatusReady {
+				if err := o.MarkForDeletion(); err != nil {
+					return err
+				}
+				o.Timestamps.Update()
+				if err := m.mutateSingleInTransaction(tx, ds.NewUpdate(m.createChunkRefKey(blob.Key, o.Key), o)); err != nil {
+					return err
+				}
+			}
+		}
+		return nil
+	})
+	return err
+}
+
+// InsertChunkRef inserts a new ChunkRef object to the datastore.
+func (m *MetaDB) InsertChunkRef(ctx context.Context, chunk *chunkref.ChunkRef) error {
+	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
+		if blob, err := m.getBlobRef(ctx, tx, chunk.BlobRef); err != nil {
+			return err
+		} else {
+			if !blob.Chunked {
+				return status.Errorf(codes.FailedPrecondition, "BlobRef (%v) is not chunked", chunk.BlobRef)
+			}
+		}
+		mut := ds.NewInsert(m.createChunkRefKey(chunk.BlobRef, chunk.Key), chunk)
+		return m.mutateSingleInTransaction(tx, mut)
+	})
+	return err
+}
+
+// UpdateChunkRef updates a ChunkRef object with the new property values.
+// Returns a NotFound error if the key is not found.
+func (m *MetaDB) UpdateChunkRef(ctx context.Context, chunk *chunkref.ChunkRef) error {
+	mut := ds.NewUpdate(m.createChunkRefKey(chunk.BlobRef, chunk.Key), chunk)
+	return m.mutateSingle(ctx, mut)
+}
+
+// MarkUncommittedBlobForDeletion marks the BlobRef specified by key for deletion
+// if the current status is StatusInitializing.
+// Returns FailedPrecondition is Status is not StatusInitializing.
+func (m *MetaDB) MarkUncommittedBlobForDeletion(ctx context.Context, key uuid.UUID) error {
+	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
+		blob, err := m.getBlobRef(ctx, tx, key)
+		if err != nil {
+			return err
+		}
+		if blob.Status != blobref.StatusInitializing {
+			return status.Errorf(codes.FailedPrecondition, "blob object is not in initialization state (state = %v)", blob.Status)
+		}
+		if err := blob.MarkForDeletion(); err != nil {
+			return err
+		}
+		blob.Timestamps.Update()
+		return m.mutateSingleInTransaction(tx, ds.NewUpdate(m.createBlobKey(key), blob))
+	})
+	return err
+}


Test patch:
diff --git a/internal/pkg/metadb/blobref/blobref_test.go b/internal/pkg/metadb/blobref/blobref_test.go
index 279b2e1..deef29d 100644
--- a/internal/pkg/metadb/blobref/blobref_test.go
+++ b/internal/pkg/metadb/blobref/blobref_test.go
@@ -60,13 +60,18 @@ func TestBlobRef_Save(t *testing.T) {
 			Name:  "RecordKey",
 			Value: record,
 		},
+		{
+			Name:  "Chunked",
+			Value: false,
+		},
 	}
 	actual, err := blob.Save()
 	assert.NoError(t, err)
 	if assert.NotNil(t, actual) {
 		assert.Equal(t, expected, actual[:len(expected)])
-		assert.Equal(t, 5, len(actual))
-		assert.Equal(t, "Timestamps", actual[4].Name)
+		if assert.Equal(t, 6, len(actual)) {
+			assert.Equal(t, "Timestamps", actual[5].Name)
+		}
 	}
 }
 
diff --git a/internal/pkg/metadb/blobref/chunkref/chunkref_test.go b/internal/pkg/metadb/blobref/chunkref/chunkref_test.go
index 1c429d0..7beb11e 100644
--- a/internal/pkg/metadb/blobref/chunkref/chunkref_test.go
+++ b/internal/pkg/metadb/blobref/chunkref/chunkref_test.go
@@ -19,30 +19,33 @@ import (
 
 	"cloud.google.com/go/datastore"
 	"github.com/google/uuid"
+	"github.com/googleforgames/open-saves/internal/pkg/metadb/blobref"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestChunkRef_New(t *testing.T) {
 	blobuuid := uuid.New()
-	c := New(blobuuid, 42, 123450)
+	c := New(blobuuid, 42)
 	assert.NotEqual(t, uuid.Nil, c.Key)
 	assert.Equal(t, blobuuid, c.BlobRef)
 	assert.Equal(t, int32(42), c.Number)
-	assert.Equal(t, int32(123450), c.Size)
+	assert.Equal(t, int32(0), c.Size)
+	assert.Equal(t, blobref.StatusInitializing, c.Status)
+	assert.NotEqual(t, uuid.Nil, c.Timestamps.Signature)
 }
 
 func TestChunkRef_ObjectPath(t *testing.T) {
-	c := New(uuid.Nil, 0, 0)
+	c := New(uuid.Nil, 0)
 	assert.Equal(t, c.Key.String(), c.ObjectPath())
 }
 
 func TestChunkRef_SaveLoad(t *testing.T) {
-	c := New(uuid.New(), 42, 24)
+	c := New(uuid.New(), 42)
 	ps, err := c.Save()
 	if err != nil {
 		t.Fatalf("Save returned error: %v", err)
 	}
-	assert.Len(t, ps, 3)
+	assert.Len(t, ps, 4)
 	loaded := new(ChunkRef)
 	if err := loaded.Load(ps); err != nil {
 		t.Fatalf("Load returned error: %v", err)
@@ -74,12 +77,12 @@ func TestChunkRef_LoadKey(t *testing.T) {
 func TestChunkRef_CacheKey(t *testing.T) {
 	testUUID := uuid.New()
 	assert.Equal(t, testUUID.String(), CacheKey(testUUID))
-	c := New(uuid.Nil, 0, 0)
+	c := New(uuid.Nil, 0)
 	assert.Equal(t, c.Key.String(), c.CacheKey())
 }
 
 func TestChunkRef_EncodeDecodeBytes(t *testing.T) {
-	c := New(uuid.New(), 42, 24)
+	c := New(uuid.New(), 42)
 	encoded, err := c.EncodeBytes()
 	if err != nil {
 		t.Fatalf("EncodeBytes failed with error: %v", err)
diff --git a/internal/pkg/metadb/metadb_test.go b/internal/pkg/metadb/metadb_test.go
index 107a4f4..b48555f 100644
--- a/internal/pkg/metadb/metadb_test.go
+++ b/internal/pkg/metadb/metadb_test.go
@@ -24,6 +24,7 @@ import (
 	pb "github.com/googleforgames/open-saves/api"
 	m "github.com/googleforgames/open-saves/internal/pkg/metadb"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/blobref"
+	"github.com/googleforgames/open-saves/internal/pkg/metadb/blobref/chunkref"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/metadbtest"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/record"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/store"
@@ -38,6 +39,7 @@ const (
 	storeKind          = "store"
 	recordKind         = "record"
 	blobKind           = "blob"
+	chunkKind          = "chunk"
 	timestampTestDelta = 5 * time.Second
 	testProject        = "triton-for-games-dev"
 	testNamespace      = "datastore-unittests"
@@ -96,6 +98,18 @@ func cloneRecord(r *record.Record) *record.Record {
 	return &ret
 }
 
+func blobRefKey(key uuid.UUID) *datastore.Key {
+	dsKey := datastore.NameKey(blobKind, key.String(), nil)
+	dsKey.Namespace = testNamespace
+	return dsKey
+}
+
+func chunkRefKey(blobKey, key uuid.UUID) *datastore.Key {
+	dsKey := datastore.NameKey(chunkKind, key.String(), blobRefKey(blobKey))
+	dsKey.Namespace = testNamespace
+	return dsKey
+}
+
 // setupTestStoreRecord creates a new store and inserts a record into it, then registers
 // cleanup functions to delete these test store and record.
 // Passing a nil to record will skip the record insertion.
@@ -130,17 +144,60 @@ func setupTestBlobRef(ctx context.Context, t *testing.T, metaDB *m.MetaDB, blob
 		t.Fatalf("InsertBlobRef failed: %v", err)
 	}
 	metadbtest.AssertEqualBlobRef(t, blob, newBlob)
-	// save the current namespace as it might change between now and when we delete
-	namespace := metaDB.Namespace
+
 	t.Cleanup(func() {
 		// Call the Datastore method directly to avoid Status checking
-		key := datastore.NameKey(blobKind, blob.Key.String(), nil)
-		key.Namespace = namespace
+		key := blobRefKey(blob.Key)
 		newDatastoreClient(ctx, t).Delete(ctx, key)
 	})
 	return newBlob
 }
 
+func setupTestChunkRef(ctx context.Context, t *testing.T, metaDB *m.MetaDB, chunk *chunkref.ChunkRef) {
+	t.Helper()
+	if err := metaDB.InsertChunkRef(ctx, chunk); err != nil {
+		t.Fatalf("InsertChunkRef failed for chunk key (%v): %v", chunk.Key, err)
+	}
+
+	t.Cleanup(func() {
+		// Call the Datastore functions directly to avoid Status checks
+		newDatastoreClient(ctx, t).Delete(ctx, chunkRefKey(chunk.BlobRef, chunk.Key))
+	})
+}
+
+func setupTestStoreRecordBlobSet(ctx context.Context, t *testing.T, metaDB *m.MetaDB, chunked bool) (*store.Store, *record.Record, *blobref.BlobRef) {
+	t.Helper()
+
+	storeKey := newStoreKey()
+	store := &store.Store{
+		Key:  storeKey,
+		Name: t.Name(),
+	}
+
+	recordKey := newRecordKey()
+	record := &record.Record{
+		Key:          recordKey,
+		OpaqueString: t.Name(),
+		Properties:   make(record.PropertyMap),
+	}
+	setupTestStoreRecord(ctx, t, metaDB, store, record)
+
+	blobKey := uuid.New()
+	blob := &blobref.BlobRef{
+		Key:       blobKey,
+		Status:    blobref.StatusInitializing,
+		StoreKey:  storeKey,
+		RecordKey: recordKey,
+		Chunked:   chunked,
+		Timestamps: timestamps.Timestamps{
+			CreatedAt: time.Unix(123, 0),
+			UpdatedAt: time.Unix(123, 0),
+		},
+	}
+	setupTestBlobRef(ctx, t, metaDB, blob)
+	return store, record, blob
+}
+
 func TestMetaDB_Disconnect(t *testing.T) {
 	ctx := context.Background()
 	metaDB := newMetaDB(ctx, t)
@@ -726,8 +783,7 @@ func TestMetaDB_ListBlobsByStatus(t *testing.T) {
 
 		// Update the timestamps here because MetaDB automatically sets UpdatesAt
 		blob.Timestamps.UpdatedAt = time.Date(2000, 1, i, 0, 0, 0, 0, time.UTC)
-		bKey := datastore.NameKey(blobKind, blob.Key.String(), nil)
-		bKey.Namespace = metaDB.Namespace
+		bKey := blobRefKey(blob.Key)
 		if _, err := client.Put(ctx, bKey, blob); err != nil {
 			t.Fatalf("Failed to change UpdatedAt for %v: %v", blob.Key, err)
 		}
@@ -839,3 +895,183 @@ func TestMetaDB_DeleteRecordWithNonExistentBlobRef(t *testing.T) {
 			"GetRecord should return NotFound after DeleteRecord")
 	}
 }
+
+func TestMetaDB_SimpleCreateGetDeleteChunkedBlob(t *testing.T) {
+	const (
+		testChunkCount = 3
+		testChunKSize  = int32(42)
+	)
+
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+	ds := newDatastoreClient(ctx, t)
+
+	store, record, blob := setupTestStoreRecordBlobSet(ctx, t, metaDB, true)
+
+	// Create Chunks with Initializing state
+	chunks := []*chunkref.ChunkRef{}
+	for i := 0; i < testChunkCount; i++ {
+		chunk := chunkref.New(blob.Key, int32(i))
+		setupTestChunkRef(ctx, t, metaDB, chunk)
+		chunks = append(chunks, chunk)
+	}
+
+	// Mark the chunks ready
+	for _, chunk := range chunks {
+		chunk.Size = testChunKSize
+		now := time.Now()
+		if err := metaDB.MarkChunkRefReady(ctx, chunk); err != nil {
+			t.Fatalf("MarkChunkRefReady failed for chunk (%v): %v", chunk.Key, err)
+		}
+		dsKey := chunkRefKey(chunk.BlobRef, chunk.Key)
+		got := new(chunkref.ChunkRef)
+		if err := ds.Get(ctx, dsKey, got); err != nil {
+			t.Errorf("Failed to get updated ChunkRef (%v): %v", dsKey, err)
+		} else {
+			assert.True(t, now.Before(got.Timestamps.UpdatedAt))
+			assert.Equal(t, blobref.StatusReady, got.Status)
+		}
+	}
+	_, blobRetrieved, err := metaDB.PromoteBlobRefToCurrent(ctx, blob)
+	if err != nil {
+		t.Fatalf("PromoteBlobRefToCurrent failed: %v", err)
+	}
+	if assert.NotNil(t, blobRetrieved) {
+		assert.EqualValues(t, testChunKSize*testChunkCount, blobRetrieved.Size)
+	}
+
+	// Verify blob and chunk metadata
+	if blobRetrieved, err := metaDB.GetCurrentBlobRef(ctx, store.Key, record.Key); err != nil {
+		t.Errorf("GetCurrentBlobRef failed: %v", err)
+	} else {
+		assert.EqualValues(t, testChunKSize*testChunkCount, blobRetrieved.Size)
+		assert.Equal(t, blob.Key, blobRetrieved.Key)
+		assert.True(t, blobRetrieved.Chunked)
+	}
+	for i := 0; i < testChunkCount; i++ {
+		got, err := metaDB.FindChunkRefByNumber(ctx, store.Key, record.Key, int32(i))
+		if assert.NoError(t, err, "FindChunkRefByNumber should not return error") {
+			metadbtest.AssertEqualChunkRef(t, chunks[i], got)
+		}
+	}
+
+	// Delete chunks and blob
+	if _, _, err := metaDB.RemoveBlobFromRecord(ctx, store.Key, record.Key); err != nil {
+		t.Fatalf("RemoveBlobFromRecord should work with Chunked blobs")
+	}
+
+	// Check if child chunks are marked as well
+	for _, chunk := range chunks {
+		got := new(chunkref.ChunkRef)
+		if err := ds.Get(ctx, chunkRefKey(chunk.BlobRef, chunk.Key), got); err != nil {
+			t.Errorf("Couldn't get chunk (%v) after RemoveBlobFromRecord: %v", chunk.Key, err)
+		} else {
+			assert.Equal(t, chunk.Key, got.Key)
+			assert.True(t, got.Timestamps.UpdatedAt.After(chunk.Timestamps.UpdatedAt))
+			assert.Equal(t, blobref.StatusPendingDeletion, got.Status)
+		}
+	}
+
+	if err := metaDB.DeleteBlobRef(ctx, blob.Key); err != nil {
+		t.Fatalf("DeleteBlobRef failed: %v", err)
+	}
+
+	for _, chunk := range chunks {
+		dsKey := chunkRefKey(chunk.BlobRef, chunk.Key)
+		got := new(chunkref.ChunkRef)
+		if err := ds.Get(ctx, dsKey, got); err != datastore.ErrNoSuchEntity {
+			t.Errorf("ChunkRef was not deleted for key (%v): %v", chunk.Key, err)
+		}
+	}
+}
+
+func TestMetaDB_MarkUncommittedBlobForDeletion(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+
+	store, record, blob := setupTestStoreRecordBlobSet(ctx, t, metaDB, true)
+
+	chunk := chunkref.New(blob.Key, 0)
+	setupTestChunkRef(ctx, t, metaDB, chunk)
+
+	assert.NoError(t, metaDB.MarkUncommittedBlobForDeletion(ctx, blob.Key))
+
+	// Should fail if blob is live
+	blob = blobref.NewChunkedBlobRef(store.Key, record.Key)
+	setupTestBlobRef(ctx, t, metaDB, blob)
+	_, _, err := metaDB.PromoteBlobRefToCurrent(ctx, blob)
+	if assert.NoError(t, err) {
+		err = metaDB.MarkUncommittedBlobForDeletion(ctx, blob.Key)
+		assert.Equal(t, codes.FailedPrecondition, status.Code(err))
+	}
+
+	// Should fail if not found
+	err = metaDB.MarkUncommittedBlobForDeletion(ctx, uuid.New())
+	assert.Equal(t, codes.NotFound, status.Code(err))
+}
+
+func TestMetaDB_UpdateChunkRef(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+
+	_, _, blob := setupTestStoreRecordBlobSet(ctx, t, metaDB, true)
+
+	chunk := chunkref.New(blob.Key, 0)
+	setupTestChunkRef(ctx, t, metaDB, chunk)
+
+	chunk.Fail()
+	assert.NoError(t, metaDB.UpdateChunkRef(ctx, chunk))
+
+	ds := newDatastoreClient(ctx, t)
+	got := new(chunkref.ChunkRef)
+	err := ds.Get(ctx, chunkRefKey(chunk.BlobRef, chunk.Key), got)
+	if assert.NoError(t, err) {
+		metadbtest.AssertEqualChunkRef(t, chunk, got)
+	}
+}
+
+func TestMetaDB_MultipleChunksWithSameNumber(t *testing.T) {
+	ctx := context.Background()
+	metaDB := newMetaDB(ctx, t)
+
+	_, _, blob := setupTestStoreRecordBlobSet(ctx, t, metaDB, true)
+
+	chunks := []*chunkref.ChunkRef{
+		chunkref.New(blob.Key, 0),
+		chunkref.New(blob.Key, 0),
+	}
+
+	for _, chunk := range chunks {
+		setupTestChunkRef(ctx, t, metaDB, chunk)
+	}
+
+	for i, chunk := range chunks {
+		chunk.Size = int32(i)
+		assert.NoError(t, metaDB.MarkChunkRefReady(ctx, chunk))
+	}
+
+	chunks[0].Status = blobref.StatusPendingDeletion
+	chunks[1].Status = blobref.StatusReady
+	ds := newDatastoreClient(ctx, t)
+	for _, chunk := range chunks {
+		got := new(chunkref.ChunkRef)
+		if err := ds.Get(ctx, chunkRefKey(chunk.BlobRef, chunk.Key), got); err != nil {
+			t.Errorf("Couldn't get ChunkRef (%v): %v", chunk.Key, err)
+		} else {
+			assert.Equal(t, chunk.Key, got.Key)
+			assert.Equal(t, chunk.Status, got.Status)
+		}
+	}
+
+	if _, blob, err := metaDB.PromoteBlobRefToCurrent(ctx, blob); err != nil {
+		t.Errorf("PromoteBlobRefToCurrent failed for BlobRef (%v): %v", blob.Key, err)
+	} else {
+		assert.EqualValues(t, chunks[1].Size, blob.Size)
+	}
+
+	if got, err := metaDB.FindChunkRefByNumber(ctx, blob.StoreKey, blob.RecordKey, 0); err != nil {
+		t.Errorf("FindChunkRefByNumber failed for BlobRef (%v): %v", blob.Key, err)
+	} else {
+		metadbtest.AssertEqualChunkRef(t, chunks[1], got)
+	}
+}
diff --git a/internal/pkg/metadb/metadbtest/metadbtest.go b/internal/pkg/metadb/metadbtest/metadbtest.go
index eb98385..a5c6f44 100644
--- a/internal/pkg/metadb/metadbtest/metadbtest.go
+++ b/internal/pkg/metadb/metadbtest/metadbtest.go
@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/blobref"
+	"github.com/googleforgames/open-saves/internal/pkg/metadb/blobref/chunkref"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/record"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/store"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/timestamps"
@@ -89,7 +90,7 @@ func AssertEqualBlobRef(t *testing.T, expected, actual *blobref.BlobRef, msgAndA
 	AssertEqualBlobRefWithinDuration(t, expected, actual, time.Duration(0), msgAndArgs...)
 }
 
-// AssertEqualBlobRefWithinDuration compares each field in metadb.BlobRef and asserts the timestamps
+// AssertEqualBlobRefWithinDuration compares each field in blobref.BlobRef and asserts the timestamps
 // are within delta.
 func AssertEqualBlobRefWithinDuration(t *testing.T, expected, actual *blobref.BlobRef, delta time.Duration, msgAndArgs ...interface{}) {
 	t.Helper()
@@ -104,6 +105,32 @@ func AssertEqualBlobRefWithinDuration(t *testing.T, expected, actual *blobref.Bl
 		assert.Equal(t, expected.Status, actual.Status, msgAndArgs...)
 		assert.Equal(t, expected.StoreKey, actual.StoreKey, msgAndArgs...)
 		assert.Equal(t, expected.RecordKey, actual.RecordKey, msgAndArgs...)
+		assert.Equal(t, expected.Chunked, actual.Chunked, msgAndArgs...)
+		assertTimestampsWithinDuration(t, &expected.Timestamps, &actual.Timestamps, delta, msgAndArgs...)
+	}
+}
+
+// AssertEqualChunkRef is equivalent to
+// AssertEqualChunkRefWithinDuration(t, expected, actual, time.Duration(0), msgAndArgs...)
+func AssertEqualChunkRef(t *testing.T, expected, actual *chunkref.ChunkRef, msgAndArgs ...interface{}) {
+	t.Helper()
+	AssertEqualChunkRefWithinDuration(t, expected, actual, time.Duration(0), msgAndArgs...)
+}
+
+// AssertEqualChunkRefWithinDuration compares each field in chunkref.ChunkRef and asserts they are equal
+// (timestamps are within delta).
+func AssertEqualChunkRefWithinDuration(t *testing.T, expected, actual *chunkref.ChunkRef, delta time.Duration, msgAndArgs ...interface{}) {
+	t.Helper()
+	if expected == nil {
+		assert.Nil(t, actual)
+		return
+	}
+	if assert.NotNil(t, actual) {
+		assert.Equal(t, expected.BlobRef, actual.BlobRef, msgAndArgs...)
+		assert.Equal(t, expected.Key, actual.Key, msgAndArgs...)
+		assert.Equal(t, expected.Number, actual.Number, msgAndArgs...)
+		assert.Equal(t, expected.Size, actual.Size, msgAndArgs...)
+		assert.Equal(t, expected.Status, actual.Status, msgAndArgs...)
 		assertTimestampsWithinDuration(t, &expected.Timestamps, &actual.Timestamps, delta, msgAndArgs...)
 	}
 }
diff --git a/internal/pkg/metadb/metadbtest/metadbtest_test.go b/internal/pkg/metadb/metadbtest/metadbtest_test.go
index 00f152b..0ea14d9 100644
--- a/internal/pkg/metadb/metadbtest/metadbtest_test.go
+++ b/internal/pkg/metadb/metadbtest/metadbtest_test.go
@@ -21,6 +21,7 @@ import (
 	"github.com/google/uuid"
 	pb "github.com/googleforgames/open-saves/api"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/blobref"
+	"github.com/googleforgames/open-saves/internal/pkg/metadb/blobref/chunkref"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/record"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/store"
 	"github.com/googleforgames/open-saves/internal/pkg/metadb/timestamps"
@@ -102,3 +103,24 @@ func TestMetaDBTest_TestAssertEqualBlobRef(t *testing.T) {
 	copied.Timestamps.UpdatedAt = original.Timestamps.UpdatedAt.Add(time.Duration(42))
 	AssertEqualBlobRefWithinDuration(t, original, copied, time.Duration(42))
 }
+
+func TestMetaDBTest_TestAssertEqualChunkRef(t *testing.T) {
+	original := &chunkref.ChunkRef{
+		Key:     uuid.MustParse("DEF55B54-F5CE-439C-9639-9F53104B9D0B"),
+		BlobRef: uuid.MustParse("E9C53FEB-2FB2-41DA-9465-81B2F59710CE"),
+		Status:  blobref.StatusError,
+		Number:  42,
+		Size:    24,
+		Timestamps: timestamps.Timestamps{
+			CreatedAt: time.Unix(100, 0),
+			UpdatedAt: time.Unix(200, 0),
+			Signature: uuid.MustParse("E871D0CC-2D83-4448-952D-9DF1F42BCB26"),
+		},
+	}
+	AssertEqualChunkRef(t, original, original)
+	copied := new(chunkref.ChunkRef)
+	*copied = *original
+	copied.Timestamps.CreatedAt = original.Timestamps.CreatedAt.Add(time.Duration(42))
+	copied.Timestamps.UpdatedAt = original.Timestamps.UpdatedAt.Add(time.Duration(-42))
+	AssertEqualChunkRefWithinDuration(t, original, copied, time.Duration(42))
+}