Issue 15: Stress test corrupts volume

Checked in for GThomas(tgulacsi78)

I've also met with partial writes (pipe closed on localhost), and also
met with real corruption: after some partial write, the data size in
needle header was read as some huuuuge number, which resulted in memory
panic.

Please consider the attached patch for ensuring full writes (seeks back
to the beginning on needle append error). Hope this helps.
Tested with a small tmpfs, lot of "no space left on device" :)
(sudo umount -lf /tmp/weed; mkdir -p /tmp/weed && sudo mount -o
size=128M,mode=4777 -t tmpfs tmpfs /tmp/weed && bin/weed master
-mdir=/tmp/weed -debug=true & bin/weed volume -dir=/tmp/weed
-debug=true)

GThomas

1 files changed, 17 insertions, 0 deletions

diff --git a/weed-fs/src/pkg/storage/needle_read_write.go b/weed-fs/src/pkg/storage/needle_read_write.go
index 5198a0a12..d74aac6a1 100644
--- a/weed-fs/src/pkg/storage/needle_read_write.go
+++ b/weed-fs/src/pkg/storage/needle_read_write.go
@@ -19,6 +19,20 @@ func (n *Needle) DiskSize() uint32 {
 	return NeedleHeaderSize + n.Size + padding + NeedleChecksumSize
 }
 func (n *Needle) Append(w io.Writer, version Version) (size uint32, err error) {
+	if s, ok := w.(io.Seeker); ok {
+		if end, e := s.Seek(0, 1); e == nil {
+			defer func(s io.Seeker, off int64) {
+				if err != nil {
+					if _, e = s.Seek(off, 0); e != nil {
+						fmt.Printf("Failed to seek back to %d with error: %s\n", w, off, e)
+					}
+				}
+			}(s, end)
+		} else {
+			err = fmt.Errorf("Cnnot Read Current Volume Position: %s", e)
+			return
+		}
+	}
 	switch version {
 	case Version1:
 		header := make([]byte, NeedleHeaderSize)
@@ -180,6 +194,9 @@ func ReadNeedleHeader(r *os.File, version Version) (n *Needle, bodyLength uint32
 //n should be a needle already read the header
 //the input stream will read until next file entry
 func (n *Needle) ReadNeedleBody(r *os.File, version Version, bodyLength uint32) (err error) {
+	if bodyLength <= 0 {
+		return nil
+	}
 	switch version {
 	case Version1:
 		bytes := make([]byte, bodyLength)