Fix tombstone overwrite bug

Author: kolesnikovae

pkg/experiment/metastore/index/tombstones/store/tombstone_store.go

@@ -17,11 +17,26 @@ var ErrInvalidTombstoneEntry = errors.New("invalid tombstone entry")
 var tombstoneBucketName = []byte("tombstones")
 
 type TombstoneEntry struct {
+	// Key the entry was stored under. This is needed for backward
+	// compatibility: if the logic for generating keys changes, we
+	// will still be able to delete old entries.
+	key        []byte
 	Index      uint64
 	AppendedAt int64
 	*metastorev1.Tombstones
 }
 
+func (e TombstoneEntry) Name() string {
+	switch {
+	case e.Tombstones.Blocks != nil:
+		return e.Tombstones.Blocks.Name
+	case e.Tombstones.Shard != nil:
+		return e.Tombstones.Shard.Name
+	default:
+		return ""
+	}
+}
+
 type TombstoneStore struct{ bucketName []byte }
 
 func NewTombstoneStore() *TombstoneStore {
@@ -83,20 +98,29 @@ func marshalTombstoneEntry(e TombstoneEntry) store.KV {
 }
 
 func marshalTombstoneEntryKey(e TombstoneEntry) []byte {
-	b := make([]byte, 16)
+	if e.key != nil {
+		b := make([]byte, len(e.key))
+		copy(b, e.key)
+		return b
+	}
+	name := e.Name()
+	b := make([]byte, 16+len(name))
 	binary.BigEndian.PutUint64(b[0:8], e.Index)
 	binary.BigEndian.PutUint64(b[8:16], uint64(e.AppendedAt))
+	copy(b[16:], name)
 	return b
 }
 
-func unmarshalTombstoneEntry(dst *TombstoneEntry, e store.KV) error {
-	if len(e.Key) < 16 {
+func unmarshalTombstoneEntry(e *TombstoneEntry, kv store.KV) error {
+	if len(kv.Key) < 16 {
 		return ErrInvalidTombstoneEntry
 	}
-	dst.Index = binary.BigEndian.Uint64(e.Key[0:8])
-	dst.AppendedAt = int64(binary.BigEndian.Uint64(e.Key[8:16]))
-	dst.Tombstones = new(metastorev1.Tombstones)
-	if err := dst.Tombstones.UnmarshalVT(e.Value); err != nil {
+	e.key = make([]byte, len(kv.Key))
+	copy(e.key, kv.Key)
+	e.Index = binary.BigEndian.Uint64(kv.Key[0:8])
+	e.AppendedAt = int64(binary.BigEndian.Uint64(kv.Key[8:16]))
+	e.Tombstones = new(metastorev1.Tombstones)
+	if err := e.Tombstones.UnmarshalVT(kv.Value); err != nil {
 		return fmt.Errorf("%w: %w", ErrInvalidTombstoneEntry, err)
 	}
 	return nil

pkg/experiment/metastore/index/tombstones/store/tombstone_store_test.go

@@ -41,7 +41,12 @@ func TestBlockQueueStore_StoreEntry(t *testing.T) {
 	var i int
 	for iter.Next() {
 		assert.Less(t, i, len(entries))
-		assert.Equal(t, entries[i], iter.At())
+		actual := iter.At()
+		expected := entries[i]
+		assert.Equal(t, expected.Index, actual.Index)
+		assert.Equal(t, expected.AppendedAt, actual.AppendedAt)
+		assert.Equal(t, expected.Tombstones, actual.Tombstones)
+		assert.NotNil(t, actual.key)
 		i++
 	}
 	assert.Nil(t, iter.Err())
@@ -92,7 +97,12 @@ func TestTombstoneStore_DeleteQueuedEntries(t *testing.T) {
 			i++
 		}
 		assert.Less(t, i, len(entries))
-		assert.Equal(t, entries[i], iter.At())
+		actual := iter.At()
+		expected := entries[i]
+		assert.Equal(t, expected.Index, actual.Index)
+		assert.Equal(t, expected.AppendedAt, actual.AppendedAt)
+		assert.Equal(t, expected.Tombstones, actual.Tombstones)
+		assert.NotNil(t, actual.key)
 		i++
 	}
 	assert.Nil(t, iter.Err())

pkg/experiment/metastore/index/tombstones/tombstones_restore_test.go

@@ -24,7 +24,7 @@ func TestTombstonesRestore(t *testing.T) {
 	require.NoError(t, ts.Init(tx))
 	require.NoError(t, tx.Commit())
 
-	for i, tombstone := range []*metastorev1.Tombstones{
+	for _, tombstone := range []*metastorev1.Tombstones{
 		{
 			Blocks: &metastorev1.BlockTombstones{
 				Name:   "x-1",
@@ -52,7 +52,7 @@ func TestTombstonesRestore(t *testing.T) {
 	} {
 		tx, err := db.Begin(true)
 		require.NoError(t, err)
-		err = ts.AddTombstones(tx, &raft.Log{Index: uint64(i), AppendedAt: now}, tombstone)
+		err = ts.AddTombstones(tx, &raft.Log{Index: 1, AppendedAt: now}, tombstone)
 		require.NoError(t, err)
 		require.NoError(t, tx.Commit())
 	}