Retry S3 chunk validation failures and preserve the underlying error

chunk_test.go

@@ -0,0 +1,50 @@
+package desync
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewChunkFromStorage(t *testing.T) {
+	conv := Converters{Compressor{}}
+	plain := []byte("the quick brown fox jumps over the lazy dog")
+	id := ChunkID(Digest.Sum(plain))
+	storage, err := conv.toStorage(plain)
+	require.NoError(t, err)
+
+	t.Run("valid", func(t *testing.T) {
+		c, err := NewChunkFromStorage(id, storage, conv, false)
+		require.NoError(t, err)
+		require.Equal(t, id, c.ID())
+	})
+
+	t.Run("undecodable storage data", func(t *testing.T) {
+		// Truncated/garbage storage data fails to decompress. This must surface as
+		// ChunkInvalid (so RepairableCache and 'verify --repair' still handle it) but
+		// must also carry the underlying decode error rather than reporting a bogus
+		// "does not match its hash 0000..." mismatch.
+		_, err := NewChunkFromStorage(id, []byte("not a valid zstd stream"), conv, false)
+		var ci ChunkInvalid
+		require.ErrorAs(t, err, &ci)
+		require.Error(t, ci.Err, "expected ChunkInvalid.Err to carry the underlying decode error")
+		require.Error(t, errors.Unwrap(err), "expected the error to unwrap to the underlying decode error")
+	})
+
+	t.Run("hash mismatch", func(t *testing.T) {
+		var wrongID ChunkID
+		wrongID[0] = 0x01
+		_, err := NewChunkFromStorage(wrongID, storage, conv, false)
+		var ci ChunkInvalid
+		require.ErrorAs(t, err, &ci)
+		require.NoError(t, ci.Err, "expected ChunkInvalid.Err to be nil for a plain hash mismatch")
+		require.Equal(t, id, ci.Sum)
+	})
+
+	t.Run("skip verify", func(t *testing.T) {
+		// With skipVerify even undecodable data is accepted (no verification done).
+		_, err := NewChunkFromStorage(id, []byte("garbage"), conv, true)
+		require.NoError(t, err)
+	})
+}

s3.go

@@ -12,6 +12,7 @@ import (
 	minio "github.com/minio/minio-go/v6"
 	"github.com/minio/minio-go/v6/pkg/credentials"
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
 )
 
 var _ WriteStore = S3Store{}
@@ -123,13 +124,32 @@ retry:
 			}
 		}
 		if attempt <= s.opt.ErrorRetry {
+			obj.Close()
 			time.Sleep(time.Duration(attempt) * s.opt.ErrorRetryBaseInterval)
 			goto retry
 		}
 		// Without ListBucket perms in AWS, we get Permission Denied for a missing chunk, not 404
 		return nil, errors.Wrap(err, fmt.Sprintf("chunk %s could not be retrieved from s3 store", id))
 	}
-	return NewChunkFromStorage(id, b, s.converters, s.opt.SkipVerify)
+
+	// A short read of the chunk body (e.g. flaky transport/endpoint) can leave us
+	// with truncated data that fails to decompress or hash. Treat that the same as
+	// other transient errors and retry under the --error-retry policy.
+	chunk, err := NewChunkFromStorage(id, b, s.converters, s.opt.SkipVerify)
+	if err != nil {
+		if attempt <= s.opt.ErrorRetry {
+			Log.WithFields(logrus.Fields{
+				"chunk":   id,
+				"object":  name,
+				"attempt": attempt,
+			}).WithError(err).Info("chunk failed validation, retrying")
+			obj.Close()
+			time.Sleep(time.Duration(attempt) * s.opt.ErrorRetryBaseInterval)
+			goto retry
+		}
+		return nil, errors.Wrap(err, s.String())
+	}
+	return chunk, nil
 }
 
 // StoreChunk adds a new chunk to the store

s3_test.go

@@ -51,7 +51,16 @@ func response(request *http.Request, headers http.Header, statusCode int, body s
 	}
 }
 
-func sendObject(conn *net.TCPConn, request *http.Request, filePath string, sendRst bool) error {
+// s3ErrMode describes how the fake S3 server should mishandle an object request.
+type s3ErrMode int
+
+const (
+	s3ErrNone        s3ErrMode = iota // serve the object normally
+	s3ErrRST                          // send half the body, then force a TCP RST (transport error)
+	s3ErrCorruptBody                  // serve a complete, well-formed response whose body is truncated
+)
+
+func sendObject(conn *net.TCPConn, request *http.Request, filePath string, mode s3ErrMode) error {
 	file, err := os.Open(filePath)
 	if err != nil {
 		if os.IsNotExist(err) {
@@ -74,19 +83,10 @@ func sendObject(conn *net.TCPConn, request *http.Request, filePath string, sendR
 	headers := http.Header{}
 	headers.Add("Last-Modified", stat.ModTime().Format(http.TimeFormat))
 	headers.Add("Content-Type", "application/octet-stream")
-	headers.Add("Content-Length", strconv.FormatInt(stat.Size(), 10))
 
-	if !sendRst {
-		resp := http.Response{
-			StatusCode: 200,
-			ProtoMajor: 1,
-			ProtoMinor: 0,
-			Request:    request,
-			Body:       file,
-			Header:     headers,
-		}
-		resp.Write(conn)
-	} else {
+	switch mode {
+	case s3ErrRST:
+		headers.Add("Content-Length", strconv.FormatInt(stat.Size(), 10))
 		if _, err := io.WriteString(conn, "HTTP/1.0 200 OK\r\n"); err != nil {
 			return err
 		}
@@ -107,11 +107,41 @@ func sendObject(conn *net.TCPConn, request *http.Request, filePath string, sendR
 		if err := conn.Close(); err != nil {
 			return err
 		}
+	case s3ErrCorruptBody:
+		// Serve a complete, well-formed HTTP response whose Content-Length matches
+		// the bytes actually written, but only write the first half of the file. The
+		// client reads it cleanly (no transport error), but the truncated chunk data
+		// fails to decompress/validate - the scenario reported in issue #334.
+		half := stat.Size() / 2
+		headers.Add("Content-Length", strconv.FormatInt(half, 10))
+		if _, err := io.WriteString(conn, "HTTP/1.0 200 OK\r\n"); err != nil {
+			return err
+		}
+		if err := headers.Write(conn); err != nil {
+			return err
+		}
+		if _, err := io.WriteString(conn, "\r\n"); err != nil {
+			return err
+		}
+		if _, err := io.CopyN(conn, file, half); err != nil {
+			return err
+		}
+	default:
+		headers.Add("Content-Length", strconv.FormatInt(stat.Size(), 10))
+		resp := http.Response{
+			StatusCode: 200,
+			ProtoMajor: 1,
+			ProtoMinor: 0,
+			Request:    request,
+			Body:       file,
+			Header:     headers,
+		}
+		resp.Write(conn)
 	}
 	return nil
 }
 
-func handleGetObjectRequest(conn *net.TCPConn, bucket, store string, errorTimes *int, errorTimesLimit int) error {
+func handleGetObjectRequest(conn *net.TCPConn, bucket, store string, errorMode s3ErrMode, errorTimes *int, errorTimesLimit int) error {
 	defer conn.Close()
 	objectGetMatcher := regexp.MustCompile(`^/` + bucket + `/(.+)$`)
 
@@ -123,7 +153,11 @@ func handleGetObjectRequest(conn *net.TCPConn, bucket, store string, errorTimes
 
 	matches := objectGetMatcher.FindStringSubmatch(request.URL.Path)
 	if matches != nil {
-		err = sendObject(conn, request, store+"/"+matches[1], *errorTimes < errorTimesLimit)
+		mode := s3ErrNone
+		if *errorTimes < errorTimesLimit {
+			mode = errorMode
+		}
+		err = sendObject(conn, request, store+"/"+matches[1], mode)
 		(*errorTimes)++
 	} else {
 		resp := response(request, http.Header{}, 400, "")
@@ -132,9 +166,10 @@ func handleGetObjectRequest(conn *net.TCPConn, bucket, store string, errorTimes
 	return err
 }
 
-// Run S3 server that can respond objects from `store`
-// if `errorTimesLimit` > 0 server will send RST packet `errorTimesLimit` times after sending half of file
-func getTcpS3Server(t *testing.T, group *errgroup.Group, ctx context.Context, bucket, store string, errorTimesLimit int) net.Listener {
+// Run S3 server that can respond objects from `store`. The first `errorTimesLimit`
+// object requests are mishandled according to `errorMode` (e.g. truncated body or
+// forced TCP reset); subsequent requests are served normally.
+func getTcpS3Server(t *testing.T, group *errgroup.Group, ctx context.Context, bucket, store string, errorMode s3ErrMode, errorTimesLimit int) net.Listener {
 	var errorTimes int
 	// using localhost + resolver let us work on hosts that support only ipv6 or only ipv4
 	ip, err := net.DefaultResolver.LookupIP(ctx, "ip", "localhost")
@@ -159,7 +194,7 @@ func getTcpS3Server(t *testing.T, group *errgroup.Group, ctx context.Context, bu
 				}
 				return err
 			}
-			err = handleGetObjectRequest(conn, bucket, store, &errorTimes, errorTimesLimit)
+			err = handleGetObjectRequest(conn, bucket, store, errorMode, &errorTimes, errorTimesLimit)
 			if err != nil {
 				return err
 			}
@@ -180,7 +215,7 @@ func TestS3StoreGetChunk(t *testing.T) {
 		ctx, cancel := context.WithCancel(context.Background())
 		group, gCtx := errgroup.WithContext(ctx)
 
-		ln := getTcpS3Server(t, group, ctx, bucket, "cmd/desync/testdata", 0)
+		ln := getTcpS3Server(t, group, ctx, bucket, "cmd/desync/testdata", s3ErrNone, 0)
 
 		group.Go(func() error {
 			defer cancel()
@@ -195,9 +230,11 @@ func TestS3StoreGetChunk(t *testing.T) {
 				chunk, err := store.GetChunk(chunkId)
 				if err != nil {
 					c <- err
+					return
 				}
 				if chunk.ID() != chunkId {
 					c <- fmt.Errorf("got chunk with id equal to %q, expected %q", chunk.ID(), chunkId)
+					return
 				}
 				c <- nil
 			}()
@@ -218,7 +255,7 @@ func TestS3StoreGetChunk(t *testing.T) {
 		ctx, cancel := context.WithCancel(context.Background())
 		group, gCtx := errgroup.WithContext(ctx)
 
-		ln := getTcpS3Server(t, group, ctx, bucket, "cmd/desync/testdata", 1)
+		ln := getTcpS3Server(t, group, ctx, bucket, "cmd/desync/testdata", s3ErrRST, 1)
 
 		group.Go(func() error {
 			defer cancel()
@@ -230,10 +267,11 @@ func TestS3StoreGetChunk(t *testing.T) {
 
 			c := make(chan error)
 			go func() {
-				_, err = store.GetChunk(chunkId)
+				_, err := store.GetChunk(chunkId)
 				opError := &net.OpError{}
 				if err == nil || !errors.As(err, &opError) {
-					c <- err
+					c <- fmt.Errorf("expected GetChunk to fail with a net.OpError, got %v", err)
+					return
 				}
 				c <- nil
 			}()
@@ -254,7 +292,94 @@ func TestS3StoreGetChunk(t *testing.T) {
 		ctx, cancel := context.WithCancel(context.Background())
 		group, gCtx := errgroup.WithContext(ctx)
 
-		ln := getTcpS3Server(t, group, ctx, bucket, "cmd/desync/testdata", 1)
+		ln := getTcpS3Server(t, group, ctx, bucket, "cmd/desync/testdata", s3ErrRST, 1)
+
+		group.Go(func() error {
+			defer cancel()
+			endpoint := url.URL{Scheme: "s3+http", Host: ln.Addr().String(), Path: "/" + bucket + "/blob1.store/"}
+			store, err := NewS3Store(&endpoint, credentials.New(&provider), location, StoreOptions{ErrorRetry: 1}, minio.BucketLookupAuto)
+			if err != nil {
+				return err
+			}
+
+			c := make(chan error)
+			go func() {
+				chunk, err := store.GetChunk(chunkId)
+				if err != nil {
+					c <- err
+					return
+				}
+				if chunk.ID() != chunkId {
+					c <- fmt.Errorf("got chunk with id equal to %q, expected %q", chunk.ID(), chunkId)
+					return
+				}
+				c <- nil
+			}()
+			select {
+			case <-gCtx.Done():
+				return nil
+			case err = <-c:
+				return err
+			}
+		})
+
+		require.NoError(t, group.Wait())
+	})
+
+	t.Run("corrupt_body_fail", func(t *testing.T) {
+		// Server returns a complete, well-formed response with a truncated body, so
+		// the chunk data fails to decompress/validate (issue #334). With no retries
+		// configured GetChunk() must return that error - and it must carry the
+		// underlying decode failure, not the bogus "hash 0000..." mismatch.
+		ctx, cancel := context.WithCancel(context.Background())
+		group, gCtx := errgroup.WithContext(ctx)
+
+		ln := getTcpS3Server(t, group, ctx, bucket, "cmd/desync/testdata", s3ErrCorruptBody, 1)
+
+		group.Go(func() error {
+			defer cancel()
+			endpoint := url.URL{Scheme: "s3+http", Host: ln.Addr().String(), Path: "/" + bucket + "/blob1.store/"}
+			store, err := NewS3Store(&endpoint, credentials.New(&provider), location, StoreOptions{}, minio.BucketLookupAuto)
+			if err != nil {
+				return err
+			}
+
+			c := make(chan error)
+			go func() {
+				_, err := store.GetChunk(chunkId)
+				if err == nil {
+					c <- errors.New("expected GetChunk to fail on truncated chunk body")
+					return
+				}
+				var ci ChunkInvalid
+				if !errors.As(err, &ci) {
+					c <- fmt.Errorf("expected ChunkInvalid, got %T: %v", err, err)
+					return
+				}
+				if ci.Err == nil {
+					c <- fmt.Errorf("expected ChunkInvalid to carry the underlying decode error, got %v", err)
+					return
+				}
+				c <- nil
+			}()
+			select {
+			case <-gCtx.Done():
+				return nil
+			case err = <-c:
+				return err
+			}
+		})
+
+		require.NoError(t, group.Wait())
+	})
+
+	t.Run("corrupt_body_recover", func(t *testing.T) {
+		// Same truncated-body scenario, but with retries enabled GetChunk() should
+		// retry the validation failure and succeed once a full response is served.
+		ctx, cancel := context.WithCancel(context.Background())
+		group, gCtx := errgroup.WithContext(ctx)
+
+		ln := getTcpS3Server(t, group, ctx, bucket, "cmd/desync/testdata", s3ErrCorruptBody, 1)
 
 		group.Go(func() error {
 			defer cancel()
@@ -269,9 +394,11 @@ func TestS3StoreGetChunk(t *testing.T) {
 				chunk, err := store.GetChunk(chunkId)
 				if err != nil {
 					c <- err
+					return
 				}
 				if chunk.ID() != chunkId {
 					c <- fmt.Errorf("got chunk with id equal to %q, expected %q", chunk.ID(), chunkId)
+					return
 				}
 				c <- nil
 			}()