Fix recovery writing extra delete record

akrylysov · akrylysov · commit 7d7fdff83a07 · 2020-11-29T17:50:34.000-05:00
diff --git a/compaction.go b/compaction.go
@@ -86,11 +86,8 @@ func (db *DB) compact(f *segment) (CompactionResult, error) {
 	return cr, err
 }
 
-func (db *DB) pickForCompaction() ([]*segment, error) {
-	segments, err := db.datalog.segmentsBySequenceID()
-	if err != nil {
-		return nil, err
-	}
+func (db *DB) pickForCompaction() []*segment {
+	segments := db.datalog.segmentsBySequenceID()
 	var picked []*segment
 	for i := len(segments) - 1; i >= 0; i-- {
 		seg := segments[i]
@@ -107,12 +104,12 @@ func (db *DB) pickForCompaction() ([]*segment, error) {
 		if seg.meta.DeleteRecords > 0 {
 			// Delete records can be discarded only when older files contain no put records for the corresponding keys.
 			// All files older than the file eligible for compaction have to be compacted.
-			return append(segments[:i+1], picked...), nil
+			return append(segments[:i+1], picked...)
 		}
 
 		picked = append([]*segment{seg}, picked...)
 	}
-	return picked, nil
+	return picked
 }
 
 // Compact compacts the DB. Deleted and overwritten items are discarded.
@@ -128,11 +125,8 @@ func (db *DB) Compact() (CompactionResult, error) {
 	}()
 
 	db.mu.RLock()
-	segments, err := db.pickForCompaction()
+	segments := db.pickForCompaction()
 	db.mu.RUnlock()
-	if err != nil {
-		return cr, err
-	}
 
 	for _, f := range segments {
 		fcr, err := db.compact(f)
diff --git a/datalog.go b/datalog.go
@@ -39,13 +39,18 @@ func openDatalog(opts *Options) (*datalog, error) {
 		if ext != segmentExt {
 			continue
 		}
-		seg, err := dl.openSegment(name)
+		id, seqID, err := parseSegmentName(name)
+		if err != nil {
+			return nil, err
+		}
+		seg, err := dl.openSegment(name, id, seqID)
 		if err != nil {
 			return nil, err
 		}
 		if seg.sequenceID > dl.maxSequenceID {
 			dl.maxSequenceID = seg.sequenceID
 		}
+		dl.segments[seg.id] = seg
 	}
 
 	if err := dl.swapSegment(); err != nil {
@@ -71,46 +76,33 @@ func parseSegmentName(name string) (uint16, uint64, error) {
 	return uint16(id), seqID, nil
 }
 
-func (dl *datalog) openSegment(name string) (*segment, error) {
-	id, seqID, err := parseSegmentName(name)
-	if err != nil {
-		return nil, err
-	}
+func (dl *datalog) openSegment(name string, id uint16, seqID uint64) (*segment, error) {
 	f, err := openFile(dl.opts.FileSystem, filepath.Join(dl.opts.path, name), false)
 	if err != nil {
 		return nil, err
 	}
+
 	meta := &segmentMeta{}
 	if !f.empty() {
 		metaPath := filepath.Join(dl.opts.path, name+metaExt)
 		if err := readGobFile(dl.opts.FileSystem, metaPath, &meta); err != nil {
 			logger.Printf("error reading segment meta %d: %v", id, err)
 			// TODO: rebuild meta?
 		}
-	} else {
-		dl.maxSequenceID++
-		seqID = dl.maxSequenceID
 	}
 
-	df := &segment{
+	seg := &segment{
 		file:       f,
 		id:         id,
 		sequenceID: seqID,
 		name:       name,
 		meta:       meta,
 	}
-	dl.segments[id] = df
-	return df, nil
+
+	return seg, nil
 }
 
 func (dl *datalog) nextWritableSegmentID() (uint16, uint64, error) {
-	for id, seg := range dl.segments {
-		// Pick unfilled segment.
-		if seg != nil && !seg.meta.Full {
-			dl.maxSequenceID++
-			return uint16(id), dl.maxSequenceID, nil
-		}
-	}
 	for id, seg := range dl.segments {
 		// Pick empty segment.
 		if seg == nil {
@@ -122,21 +114,29 @@ func (dl *datalog) nextWritableSegmentID() (uint16, uint64, error) {
 }
 
 func (dl *datalog) swapSegment() error {
+	// Pick unfilled segment.
+	for _, seg := range dl.segments {
+		if seg != nil && !seg.meta.Full {
+			dl.curSeg = seg
+			return nil
+		}
+	}
+
+	// Create new segment.
 	id, seqID, err := dl.nextWritableSegmentID()
 	if err != nil {
 		return err
 	}
-	var seg *segment
-	if dl.segments[id] != nil {
-		seg = dl.segments[id]
-	} else {
-		name := segmentName(id, seqID)
-		seg, err = dl.openSegment(name)
-		if err != nil {
-			return err
-		}
+
+	name := segmentName(id, seqID)
+	seg, err := dl.openSegment(name, id, seqID)
+	if err != nil {
+		return err
 	}
+
+	dl.segments[id] = seg
 	dl.curSeg = seg
+
 	return nil
 }
 
@@ -189,14 +189,13 @@ func (dl *datalog) readKey(sl slot) ([]byte, error) {
 	return key, nil*/
 }
 
-func (dl *datalog) trackOverwrite(sl slot) {
+func (dl *datalog) trackDel(sl slot) {
 	meta := dl.segments[sl.segmentID].meta
 	meta.DeletedKeys++
 	meta.DeletedBytes += encodedRecordSize(sl.kvSize())
 }
 
-func (dl *datalog) del(key []byte, sl slot) error {
-	dl.trackOverwrite(sl)
+func (dl *datalog) del(key []byte) error {
 	delRecord := encodeDeleteRecord(key)
 	_, _, err := dl.writeRecord(delRecord, recordTypeDelete)
 	if err != nil {
@@ -226,7 +225,7 @@ func (dl *datalog) writeRecord(data []byte, rt recordType) (uint16, uint32, erro
 	return dl.curSeg.id, uint32(off), nil
 }
 
-func (dl *datalog) writeKeyValue(key []byte, value []byte) (uint16, uint32, error) {
+func (dl *datalog) put(key []byte, value []byte) (uint16, uint32, error) {
 	return dl.writeRecord(encodePutRecord(key, value), recordTypePut)
 }
 
@@ -250,20 +249,20 @@ func (dl *datalog) close() error {
 	return nil
 }
 
-func (dl *datalog) segmentsBySequenceID() ([]*segment, error) {
-	// Sort segments in ascending order by sequence ID.
+// segmentsBySequenceID returns segments ordered from oldest to newest.
+func (dl *datalog) segmentsBySequenceID() []*segment {
 	var segments []*segment
 
-	for _, f := range dl.segments {
-		if f == nil {
+	for _, seg := range dl.segments {
+		if seg == nil {
 			continue
 		}
-		segments = append(segments, f)
+		segments = append(segments, seg)
 	}
 
 	sort.SliceStable(segments, func(i, j int) bool {
 		return segments[i].sequenceID < segments[j].sequenceID
 	})
 
-	return segments, nil
+	return segments
 }
diff --git a/datalog_test.go b/datalog_test.go
@@ -6,31 +6,37 @@ import (
 	"github.com/akrylysov/pogreb/internal/assert"
 )
 
+func (dl *datalog) segmentMetas() []segmentMeta {
+	var metas []segmentMeta
+	for _, seg := range dl.segmentsBySequenceID() {
+		metas = append(metas, *seg.meta)
+	}
+	return metas
+}
+
 func TestDatalog(t *testing.T) {
 	db, err := createTestDB(nil)
 	assert.Nil(t, err)
 
-	_, _, err = db.datalog.writeKeyValue([]byte{'1'}, []byte{'1'})
+	_, _, err = db.datalog.put([]byte{'1'}, []byte{'1'})
 	assert.Nil(t, err)
 	assert.Equal(t, &segmentMeta{PutRecords: 1}, db.datalog.segments[0].meta)
 	assert.Nil(t, db.datalog.segments[1])
 
-	sm, err := db.datalog.segmentsBySequenceID()
-	assert.Nil(t, err)
+	sm := db.datalog.segmentsBySequenceID()
 	assert.Equal(t, []*segment{db.datalog.segments[0]}, sm)
 
 	// Writing to a full file swaps it.
 	db.datalog.segments[0].meta.Full = true
-	_, _, err = db.datalog.writeKeyValue([]byte{'1'}, []byte{'1'})
+	_, _, err = db.datalog.put([]byte{'1'}, []byte{'1'})
 	assert.Nil(t, err)
 	assert.Equal(t, &segmentMeta{PutRecords: 1, Full: true}, db.datalog.segments[0].meta)
 	assert.Equal(t, &segmentMeta{PutRecords: 1}, db.datalog.segments[1].meta)
 
-	sm, err = db.datalog.segmentsBySequenceID()
-	assert.Nil(t, err)
+	sm = db.datalog.segmentsBySequenceID()
 	assert.Equal(t, []*segment{db.datalog.segments[0], db.datalog.segments[1]}, sm)
 
-	_, _, err = db.datalog.writeKeyValue([]byte{'1'}, []byte{'1'})
+	_, _, err = db.datalog.put([]byte{'1'}, []byte{'1'})
 	assert.Nil(t, err)
 	assert.Equal(t, &segmentMeta{PutRecords: 1, Full: true}, db.datalog.segments[0].meta)
 	assert.Equal(t, &segmentMeta{PutRecords: 2}, db.datalog.segments[1].meta)
diff --git a/db.go b/db.go
@@ -251,7 +251,7 @@ func (db *DB) put(sl slot, key []byte) error {
 			return true, err
 		}
 		if bytes.Equal(key, slKey) {
-			db.datalog.trackOverwrite(cursl) // Overwriting existing key.
+			db.datalog.trackDel(cursl) // Overwriting existing key.
 			return true, nil
 		}
 		return false, nil
@@ -271,7 +271,7 @@ func (db *DB) Put(key []byte, value []byte) error {
 	db.mu.Lock()
 	defer db.mu.Unlock()
 
-	fileID, offset, err := db.datalog.writeKeyValue(key, value)
+	fileID, offset, err := db.datalog.put(key, value)
 	if err != nil {
 		return err
 	}
@@ -294,7 +294,7 @@ func (db *DB) Put(key []byte, value []byte) error {
 	return nil
 }
 
-func (db *DB) del(h uint32, key []byte) error {
+func (db *DB) del(h uint32, key []byte, writeWAL bool) error {
 	err := db.index.delete(h, func(sl slot) (b bool, e error) {
 		if uint16(len(key)) != sl.keySize {
 			return false, nil
@@ -304,7 +304,12 @@ func (db *DB) del(h uint32, key []byte) error {
 			return true, err
 		}
 		if bytes.Equal(key, slKey) {
-			return true, db.datalog.del(key, sl)
+			db.datalog.trackDel(sl)
+			var err error
+			if writeWAL {
+				err = db.datalog.del(key)
+			}
+			return true, err
 		}
 		return false, nil
 	})
@@ -317,7 +322,7 @@ func (db *DB) Delete(key []byte) error {
 	db.metrics.Dels.Add(1)
 	db.mu.Lock()
 	defer db.mu.Unlock()
-	if err := db.del(h, key); err != nil {
+	if err := db.del(h, key, true); err != nil {
 		return err
 	}
 	if db.syncWrites {
diff --git a/db_test.go b/db_test.go
@@ -82,7 +82,7 @@ func TestEmpty(t *testing.T) {
 	assert.Nil(t, db.Close())
 }
 
-func TestSimple(t *testing.T) {
+func TestFull(t *testing.T) {
 	db, err := createTestDB(&Options{maxSegmentSize: 1024, BackgroundSyncInterval: -1})
 	assert.Nil(t, err)
 	var i byte
@@ -135,6 +135,7 @@ func TestSimple(t *testing.T) {
 		assert.Nil(t, db.Close())
 	}
 
+	expectedSegMetas := db.datalog.segmentMetas()
 	verifyKeysAndClose(0)
 
 	// Simulate crash.
@@ -147,8 +148,7 @@ func TestSimple(t *testing.T) {
 	assert.Nil(t, err)
 	verifyKeysAndClose(0)
 
-	assert.Equal(t, segmentMeta{PutRecords: 42, DeleteRecords: 1, DeletedBytes: 11}, *db.datalog.segments[0].meta)
-	assert.Equal(t, segmentMeta{PutRecords: 42}, *db.datalog.segments[1].meta)
+	assert.Equal(t, expectedSegMetas, db.datalog.segmentMetas())
 
 	// Update all items
 	db, err = Open("test.db", nil)
diff --git a/internal/assert/assert_test.go b/internal/assert/assert_test.go
@@ -67,7 +67,7 @@ func TestEqual(t *testing.T) {
 	}
 
 	for i, tc := range testCases {
-		t.Run(fmt.Sprintf("case %d %+v", i, tc), func(t *testing.T) {
+		t.Run(fmt.Sprintf("%d %+v", i, tc), func(t *testing.T) {
 			mock := &testing.T{}
 			wg := &sync.WaitGroup{}
 			wg.Add(1)
@@ -129,7 +129,7 @@ func TestNil(t *testing.T) {
 	}
 
 	for i, tc := range testCases {
-		t.Run(fmt.Sprintf("case %d %+v", i, tc.obj), func(t *testing.T) {
+		t.Run(fmt.Sprintf("%d %+v", i, tc.obj), func(t *testing.T) {
 			mock := &testing.T{}
 			wg := &sync.WaitGroup{}
 			wg.Add(1)
diff --git a/internal/hash/murmurhash32_test.go b/internal/hash/murmurhash32_test.go
@@ -44,7 +44,7 @@ func TestSum32WithSeed(t *testing.T) {
 		},
 	}
 	for i, tc := range testCases {
-		t.Run(fmt.Sprintf("case %d", i), func(t *testing.T) {
+		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
 			assert.Equal(t, tc.out, Sum32WithSeed(tc.in, tc.seed))
 		})
 	}
diff --git a/recovery.go b/recovery.go
diff --git a/recovery_test.go b/recovery_test.go

Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,7 @@ func TestSum32WithSeed(t *testing.T) {`
`44`	`44`	`},`
`45`	`45`	`}`
`46`	`46`	`for i, tc := range testCases {`
`47`		`- t.Run(fmt.Sprintf("case %d", i), func(t *testing.T) {`
	`47`	`+ t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {`
`48`	`48`	`assert.Equal(t, tc.out, Sum32WithSeed(tc.in, tc.seed))`
`49`	`49`	`})`
`50`	`50`	`}`