adds more error message

weed/storage/volume_backup.go

@@ -191,12 +191,13 @@ func (v *Volume) BinarySearchByAppendAtNs(sinceNs uint64) (offset Offset, isLast
 		// read the appendAtNs for entry m
 		offset, err = v.readOffsetFromIndex(m)
 		if err != nil {
+			err = fmt.Errorf("read entry %d: %v", m, err)
 			return
 		}
 
 		mNs, nsReadErr := v.readAppendAtNs(offset)
 		if nsReadErr != nil {
-			err = nsReadErr
+			err = fmt.Errorf("read entry %d offset %d: %v", m, offset, nsReadErr)
 			return
 		}
 

weed/storage/volume_write_test.go

@@ -0,0 +1,69 @@
+package storage
+
+import (
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/storage/needle"
+	"github.com/chrislusf/seaweedfs/weed/storage/super_block"
+	"io/ioutil"
+	"os"
+	"testing"
+	"time"
+)
+
+func TestSearchVolumesWithDeletedNeedles(t *testing.T) {
+	dir, err := ioutil.TempDir("", "example")
+	if err != nil {
+		t.Fatalf("temp dir creation: %v", err)
+	}
+	defer os.RemoveAll(dir) // clean up
+
+	v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, 0)
+	if err != nil {
+		t.Fatalf("volume creation: %v", err)
+	}
+
+	count := 10
+
+	for i:=1;i<count;i++{
+		n := newRandomNeedle(uint64(i))
+		_, _, _, err := v.writeNeedle2(n, true, false)
+		if err != nil {
+			t.Fatalf("write needle %d: %v", i, err)
+		}
+	}
+
+	for i:=1;i<5;i++{
+		n := newEmptyNeedle(uint64(i))
+		_, err := v.doDeleteRequest(n)
+		if err != nil {
+			t.Fatalf("delete needle %d: %v", i, err)
+		}
+	}
+
+	ts1 := time.Now().UnixNano()
+
+	var ts2 uint64
+
+	for i:=5;i<count;i++{
+		n := newEmptyNeedle(uint64(i))
+		_, err := v.doDeleteRequest(n)
+		if err != nil {
+			t.Fatalf("delete needle %d: %v", i, err)
+		}
+		ts2 = n.AppendAtNs
+	}
+
+	offset, isLast, err := v.BinarySearchByAppendAtNs(uint64(ts1))
+	if err != nil {
+		t.Fatalf("lookup by ts: %v", err)
+	}
+	fmt.Printf("offset: %v, isLast: %v\n", offset.ToActualOffset(), isLast)
+
+	offset, isLast, err = v.BinarySearchByAppendAtNs(uint64(ts2))
+	if err != nil {
+		t.Fatalf("lookup by ts: %v", err)
+	}
+	fmt.Printf("offset: %v, isLast: %v\n", offset.ToActualOffset(), isLast)
+
+
+}