fix!: remove indexing for timestamps

deploy/datastore/index.yaml

@@ -1,36 +1,17 @@
 indexes:
-
-- kind: blob
-  properties:
-  - name: Status
-    direction: asc
-  - name: Timestamps.UpdatedAt
-    direction: asc
-
-- kind: chunk
-  properties:
-  - name: Status
-    direction: asc
-  - name: Timestamps.UpdatedAt
-    direction: asc
+  - kind: record
+    ancestor: yes
+    properties:
+      - name: Properties.prop1
+        direction: asc
 
-- kind: record
-  ancestor: yes
-  properties:
-  - name: Properties.prop1
-    direction: asc 
+  - kind: record
+    ancestor: yes
+    properties:
+      - name: Properties.prop1
 
-- kind: record
-  ancestor: yes
-  properties:
-  - name: Properties.prop1
-
-- kind: record
-  ancestor: yes
-  properties:
-  - name: Timestamps.CreatedAt
-
-- kind: record
-  ancestor: yes
-  properties:
-  - name: Timestamps.UpdatedAt
+  - kind: record
+    ancestor: yes
+    properties:
+      - name: Properties.prop1
+        direction: desc

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/alicebob/miniredis/v2 v2.22.0
 	github.com/go-redis/redis/v8 v8.11.5
 	github.com/golang/mock v1.6.0
+	github.com/google/go-cmp v0.5.8
 	github.com/google/uuid v1.3.0
 	github.com/pseudomuto/protoc-gen-doc v1.5.1
 	github.com/sirupsen/logrus v1.9.0
@@ -38,7 +39,6 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
-	github.com/google/go-cmp v0.5.8 // indirect
 	github.com/google/wire v0.5.0 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.1.0 // indirect
 	github.com/googleapis/gax-go/v2 v2.4.0 // indirect

internal/app/collector/collector.go

@@ -172,7 +172,7 @@ func (c *Collector) deleteBlob(ctx context.Context, blob *blobref.BlobRef) {
 
 func (c *Collector) deleteMatchingBlobRefs(ctx context.Context, status blobref.Status, olderThan time.Time) error {
 	log.Infof("Garbage collecting BlobRef objects with status = %v, and older than %v", status, olderThan)
-	cursor, err := c.metaDB.ListBlobRefsByStatus(ctx, status, olderThan)
+	cursor, err := c.metaDB.ListBlobRefsByStatus(ctx, status)
 	if err != nil {
 		log.Fatalf("ListBlobRefsByStatus returned error: %v", err)
 		return err
@@ -186,14 +186,16 @@ func (c *Collector) deleteMatchingBlobRefs(ctx context.Context, status blobref.S
 			log.Errorf("cursor.Next() returned error: %v", err)
 			break
 		}
-		c.deleteBlob(ctx, blob)
+		if blob.Timestamps.UpdatedAt.Before(olderThan) {
+			c.deleteBlob(ctx, blob)
+		}
 	}
 	return nil
 }
 
 func (c *Collector) deleteMatchingChunkRefs(ctx context.Context, status blobref.Status, olderThan time.Time) error {
-	log.Infof("Garbage collecting ChunkRef objects with status = %v, older than %v", status, olderThan)
-	cursor := c.metaDB.ListChunkRefsByStatus(ctx, status, olderThan)
+	log.Infof("Garbage collecting ChunkRef objects with status = %v, and older than %v", status, olderThan)
+	cursor := c.metaDB.ListChunkRefsByStatus(ctx, status)
 	for {
 		chunk, err := cursor.Next()
 		if err == iterator.Done {
@@ -203,12 +205,14 @@ func (c *Collector) deleteMatchingChunkRefs(ctx context.Context, status blobref.
 			log.Errorf("cursor.Next() return error: %v", err)
 			return err
 		}
-		if err := c.deleteChunk(ctx, chunk); err != nil {
-			log.Errorf("deleteChunk failed for chunk (%v): %v", chunk.Key, err)
-			continue
-		}
-		if err := c.metaDB.DeleteChunkRef(ctx, chunk.BlobRef, chunk.Key); err != nil {
-			log.Errorf("DeleteChunkRef failed for chunk (%v): %v", chunk.Key, err)
+		if chunk.Timestamps.UpdatedAt.Before(olderThan) {
+			if err := c.deleteChunk(ctx, chunk); err != nil {
+				log.Errorf("deleteChunk failed for chunk (%v): %v", chunk.Key, err)
+				continue
+			}
+			if err := c.metaDB.DeleteChunkRef(ctx, chunk.BlobRef, chunk.Key); err != nil {
+				log.Errorf("DeleteChunkRef failed for chunk (%v): %v", chunk.Key, err)
+			}
 		}
 	}
 	return nil

internal/app/collector/collector_test.go

@@ -155,8 +155,8 @@ func TestCollector_DeletesBlobs(t *testing.T) {
 	// 4 is still initializing
 	for i := 0; i < blobRefCount; i++ {
 		blobRef := blobref.NewBlobRef(0, store.Key, record.Key)
-		blobRef.Timestamps.CreatedAt = collector.cfg.Before
-		blobRef.Timestamps.UpdatedAt = collector.cfg.Before
+		blobRef.Timestamps.CreatedAt = time.Now()
+		blobRef.Timestamps.UpdatedAt = time.Now()
 		blobRefs = append(blobRefs, blobRef)
 	}
 	blobRefs[0].MarkForDeletion()
@@ -242,8 +242,8 @@ func TestCollector_DeletesChunkedBlobs(t *testing.T) {
 	// 4 is still initializing
 	for i := 0; i < chunkRefCount; i++ {
 		chunk := chunkref.New(blob.Key, int32(i))
-		chunk.Timestamps.CreatedAt = collector.cfg.Before
-		chunk.Timestamps.UpdatedAt = collector.cfg.Before
+		chunk.Timestamps.CreatedAt = time.Now()
+		chunk.Timestamps.UpdatedAt = time.Now()
 		chunks = append(chunks, chunk)
 	}
 	chunks[0].MarkForDeletion()

internal/app/server/open_saves_test.go

@@ -1164,21 +1164,6 @@ func TestOpenSaves_QueryRecords_Order(t *testing.T) {
 	assert.Equal(t, resp.Records[0].Properties["prop1"].Value, intVal1)
 	assert.Equal(t, resp.Records[1].Properties["prop1"].Value, intVal2)
 
-	queryReq = &pb.QueryRecordsRequest{
-		StoreKey: storeKey,
-		SortOrders: []*pb.SortOrder{
-			{
-				Property:  pb.SortOrder_UPDATED_AT,
-				Direction: pb.SortOrder_ASC,
-			},
-		},
-	}
-	resp, err = client.QueryRecords(ctx, queryReq)
-	// These records are created at the same time so no way of verifying the order.
-	// Just make sure there's no error returned by this query.
-	require.NoError(t, err)
-	require.Equal(t, 2, len(resp.Records))
-
 	// Test errors
 	queryReq = &pb.QueryRecordsRequest{
 		StoreKey: storeKey,
@@ -1191,18 +1176,6 @@ func TestOpenSaves_QueryRecords_Order(t *testing.T) {
 	}
 	_, err = client.QueryRecords(ctx, queryReq)
 	assert.Equal(t, codes.InvalidArgument, status.Code(err))
-
-	queryReq = &pb.QueryRecordsRequest{
-		StoreKey: storeKey,
-		SortOrders: []*pb.SortOrder{
-			{
-				Property:  pb.SortOrder_CREATED_AT,
-				Direction: 3,
-			},
-		},
-	}
-	_, err = client.QueryRecords(ctx, queryReq)
-	assert.Equal(t, codes.InvalidArgument, status.Code(err))
 }
 
 func TestOpenSaves_QueryRecords_Limit(t *testing.T) {

internal/pkg/metadb/blobref/blobref_test.go

@@ -155,12 +155,14 @@ func TestBlobRef_Save(t *testing.T) {
 			Value: &datastore.Entity{
 				Properties: []datastore.Property{
 					{
-						Name:  "CreatedAt",
-						Value: time.Date(1992, 1, 15, 3, 15, 55, 0, time.UTC),
+						Name:    "CreatedAt",
+						Value:   time.Date(1992, 1, 15, 3, 15, 55, 0, time.UTC),
+						NoIndex: true,
 					},
 					{
-						Name:  "UpdatedAt",
-						Value: time.Date(1992, 11, 27, 1, 3, 11, 0, time.UTC),
+						Name:    "UpdatedAt",
+						Value:   time.Date(1992, 11, 27, 1, 3, 11, 0, time.UTC),
+						NoIndex: true,
 					},
 					{
 						Name:    "Signature",

internal/pkg/metadb/metadb.go

@@ -19,7 +19,6 @@ import (
 	"errors"
 	"fmt"
 	"strconv"
-	"time"
 
 	ds "cloud.google.com/go/datastore"
 	"github.com/google/uuid"
@@ -55,10 +54,10 @@ var (
 // The methods return gRPC error codes. Here are some common error codes
 // returned. For additional details, please look at the method help.
 // Common errors:
-//	- NotFound: entity or object is not found
-//	- Aborted: transaction is aborted
-//	- InvalidArgument: key or value provided is not valid
-//	- Internal: internal unrecoverable error
+//   - NotFound: entity or object is not found
+//   - Aborted: transaction is aborted
+//   - InvalidArgument: key or value provided is not valid
+//   - Internal: internal unrecoverable error
 type MetaDB struct {
 	// Datastore namespace for multi-tenancy
 	Namespace string
@@ -417,7 +416,7 @@ func (m *MetaDB) UpdateBlobRef(ctx context.Context, blob *blobref.BlobRef) (*blo
 
 // GetBlobRef returns a BlobRef object specified by the key.
 // Returns errors:
-//	- NotFound: the object is not found.
+//   - NotFound: the object is not found.
 func (m *MetaDB) GetBlobRef(ctx context.Context, key uuid.UUID) (*blobref.BlobRef, error) {
 	return m.getBlobRef(ctx, nil, key)
 }
@@ -434,8 +433,8 @@ func (m *MetaDB) getCurrentBlobRef(ctx context.Context, tx *ds.Transaction, stor
 
 // GetCurrentBlobRef gets a BlobRef object associated with a record.
 // Returned errors:
-// 	- NotFound: the record is not found.
-// 	- FailedPrecondition: the record doesn't have a blob.
+//   - NotFound: the record is not found.
+//   - FailedPrecondition: the record doesn't have a blob.
 func (m *MetaDB) GetCurrentBlobRef(ctx context.Context, storeKey, recordKey string) (*blobref.BlobRef, error) {
 	var blob *blobref.BlobRef
 	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
@@ -480,8 +479,8 @@ func (m *MetaDB) chunkObjectsSizeSum(ctx context.Context, tx *ds.Transaction, bl
 // PromoteBlobRefToCurrent promotes the provided BlobRef object as a current
 // external blob reference.
 // Returned errors:
-//	- NotFound: the specified record or the blobref was not found
-//  - Internal: BlobRef status transition error
+//   - NotFound: the specified record or the blobref was not found
+//   - Internal: BlobRef status transition error
 func (m *MetaDB) PromoteBlobRefToCurrent(ctx context.Context, blob *blobref.BlobRef) (*record.Record, *blobref.BlobRef, error) {
 	record := new(record.Record)
 	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
@@ -550,9 +549,9 @@ func (m *MetaDB) PromoteBlobRefToCurrent(ctx context.Context, blob *blobref.Blob
 // storeKey and recordKey. It also changes the status of the blob object to
 // BlobRefStatusPendingDeletion.
 // Returned errors:
-//	- NotFound: the specified record or the blobref was not found
-//	- FailedPrecondition: the record doesn't have an external blob
-//  - Internal: BlobRef status transition error
+//   - NotFound: the specified record or the blobref was not found
+//   - FailedPrecondition: the record doesn't have an external blob
+//   - Internal: BlobRef status transition error
 func (m *MetaDB) RemoveBlobFromRecord(ctx context.Context, storeKey string, recordKey string) (*record.Record, *blobref.BlobRef, error) {
 	rkey := m.createRecordKey(storeKey, recordKey)
 	blob := new(blobref.BlobRef)
@@ -629,8 +628,8 @@ func (m *MetaDB) deleteChildChunkRefs(ctx context.Context, tx *ds.Transaction, b
 
 // DeleteBlobRef deletes the BlobRef object from the database.
 // Returned errors:
-//	- NotFound: the blobref object is not found
-//	- FailedPrecondition: the blobref status is Ready and can't be deleted
+//   - NotFound: the blobref object is not found
+//   - FailedPrecondition: the blobref status is Ready and can't be deleted
 func (m *MetaDB) DeleteBlobRef(ctx context.Context, key uuid.UUID) error {
 	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
 		blob, err := m.getBlobRef(ctx, tx, key)
@@ -652,8 +651,8 @@ func (m *MetaDB) DeleteBlobRef(ctx context.Context, key uuid.UUID) error {
 
 // DeleteChunkRef deletes the ChunkRef object from the database.
 // Returned errors:
-//	- NotFound: the chunkref object is not found.
-//	- FailedPrecondition: the chunkref status is Ready and can't be deleted.
+//   - NotFound: the chunkref object is not found.
+//   - FailedPrecondition: the chunkref status is Ready and can't be deleted.
 func (m *MetaDB) DeleteChunkRef(ctx context.Context, blobKey, key uuid.UUID) error {
 	_, err := m.client.RunInTransaction(ctx, func(tx *ds.Transaction) error {
 		var chunk chunkref.ChunkRef
@@ -669,19 +668,17 @@ func (m *MetaDB) DeleteChunkRef(ctx context.Context, blobKey, key uuid.UUID) err
 }
 
 // ListBlobRefsByStatus returns a cursor that iterates over BlobRefs
-// where Status = status and UpdatedAt < olderThan.
-func (m *MetaDB) ListBlobRefsByStatus(ctx context.Context, status blobref.Status, olderThan time.Time) (*blobref.BlobRefCursor, error) {
-	query := m.newQuery(blobKind).Filter("Status = ", int(status)).
-		Filter("Timestamps.UpdatedAt <", olderThan)
+// where Status = status.
+func (m *MetaDB) ListBlobRefsByStatus(ctx context.Context, status blobref.Status) (*blobref.BlobRefCursor, error) {
+	query := m.newQuery(blobKind).Filter("Status = ", int(status))
 	iter := blobref.NewCursor(m.client.Run(ctx, query))
 	return iter, nil
 }
 
 // ListChunkRefsByStatus returns a cursor that iterates over ChunkRefs
-// where Status = status and UpdatedAt < olderThan.
-func (m *MetaDB) ListChunkRefsByStatus(ctx context.Context, status blobref.Status, olderThan time.Time) *chunkref.ChunkRefCursor {
-	query := m.newQuery(chunkKind).Filter("Status = ", int(status)).
-		Filter("Timestamps.UpdatedAt <", olderThan)
+// where Status = status.
+func (m *MetaDB) ListChunkRefsByStatus(ctx context.Context, status blobref.Status) *chunkref.ChunkRefCursor {
+	query := m.newQuery(chunkKind).Filter("Status = ", int(status))
 	return chunkref.NewCursor(m.client.Run(ctx, query))
 }