Merge pull request #15 from Preetam/fix-races-snapshot-reads

Fix race conditions, snapshot read bugs

cursor.go

@@ -65,7 +65,7 @@ func (c *Cursor) Next() bool {
 
 	c.current.lock.RLock()
 	for (c.current.Deleted != 0 && c.current.Deleted <= c.snapshot) ||
-		(c.current.Offset > c.snapshot) {
+		(c.current.Offset >= c.snapshot) {
 		rec, err = c.collection.readRecord(c.current.Next)
 		if err != nil {
 			c.current.lock.RUnlock()
@@ -100,33 +100,38 @@ func (c *Cursor) Value() string {
 }
 
 // Seek positions the cursor at the last key less than
-// the provided key.
+// or equal to the provided key.
 func (c *Cursor) Seek(key string) {
+	var rec *record
+	var err error
 	offset := c.collection.cache.findLastLessThan(key)
 	if offset == 0 {
-		head, err := c.collection.readRecord(c.collection.Head)
+		c.collection.metaLock.RLock()
+		rec, err = c.collection.readRecord(c.collection.Head)
+		c.collection.metaLock.RUnlock()
+		if err != nil {
+			c.current = nil
+			return
+		}
+	} else {
+		rec, err = c.collection.readRecord(offset)
 		if err != nil {
 			c.current = nil
 			return
 		}
-		c.current = head
-		c.first = true
-		return
-	}
-	rec, err := c.collection.readRecord(offset)
-	if err != nil {
-		c.current = nil
-		return
 	}
 	c.current = rec
 	c.first = true
 	for rec != nil {
 		rec.lock.RLock()
 		if rec.Key > key {
+			rec.lock.RUnlock()
 			break
 		}
-		if (rec.Deleted > 0 && rec.Deleted < c.snapshot) || (c.current.Offset > c.snapshot) {
-			rec.lock.RUnlock()
+		if (rec.Deleted > 0 && rec.Deleted <= c.snapshot) || (rec.Offset >= c.snapshot) {
+			oldRec := rec
+			rec = c.collection.nextRecord(rec)
+			oldRec.lock.RUnlock()
 			continue
 		}
 		if rec.Key <= key {

lm2_test.go

@@ -473,3 +473,54 @@ func TestSeekToFirstKey(t *testing.T) {
 		t.Fatalf("expected cursor key to be 'a', got %v", cur.Key())
 	}
 }
+
+func TestOverwriteFirstKey(t *testing.T) {
+	c, err := NewCollection("/tmp/test_overwritefirstkey.lm2", 100)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer c.Close()
+
+	wb := NewWriteBatch()
+	wb.Set("a", "1")
+	wb.Set("b", "1")
+	wb.Set("c", "1")
+	wb.Set("d", "1")
+	_, err = c.Update(wb)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	wb = NewWriteBatch()
+	wb.Set("a", "2")
+	_, err = c.Update(wb)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	cur, err := c.NewCursor()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	cur.Seek("a")
+	if !cur.Valid() {
+		t.Fatal("expected cursor to be valid")
+	}
+
+	if !cur.Next() {
+		t.Fatal("expected Next() to return true")
+	}
+
+	if cur.Key() != "a" {
+		t.Fatalf("expected cursor key to be 'a', got %v", cur.Key())
+	}
+
+	if !cur.Next() {
+		t.Fatal("expected Next() to return true")
+	}
+
+	if cur.Key() != "b" {
+		t.Fatalf("expected cursor key to be 'b', got %v", cur.Key())
+	}
+}