Fix: Restart S3 reads from beginning after errors

README.md

@@ -2,6 +2,16 @@
 
 yproxy - a service for efficient transfer data external storages.
 
+## Retrying interrupted reads
+
+The `storage.read_retry_mode` setting controls how yproxy resumes an S3 read
+after a transient error. The default `resume` retries with an S3 Range request.
+Set `restart_from_beginning` to reopen the object at offset zero, recreate the
+decryption reader, and discard plaintext already sent to the client. This mode is
+slower after an error but avoids relying on a nonzero Range response.
+
+`storage.read_retry_attempts` limits full restarts and defaults to `3`.
+
 ## vacuum / garbage collection configuration
 
 The `vacuum` section of the yproxy configuration file controls the

config/instance_test.go

@@ -42,7 +42,7 @@
 }
 
 func TestReadInstanceConfigPreservesExplicitZeroPortsYAML(t *testing.T) {
 	cfg, err := ReadInstanceConfig(writeTestConfig(t, "yproxy.yaml", "stat_port: 0\npsql_port: 0\nmetrics_port: 0\n"))
 	if err != nil {
 		t.Fatalf("failed to read config: %v", err)
 	}
@@ -168,6 +168,12 @@
 	if storage.EndpointSourceScheme != DefaultEndpointSourceScheme {
 		t.Fatalf("expected default endpoint source scheme %q, got %q", DefaultEndpointSourceScheme, storage.EndpointSourceScheme)
 	}
+	if storage.ReadRetryMode != DefaultReadRetryMode {
+		t.Fatalf("expected default read retry mode %q, got %q", DefaultReadRetryMode, storage.ReadRetryMode)
+	}
+	if storage.ReadRetryAttempts != DefaultReadRetryAttempts {
+		t.Fatalf("expected default read retry attempts %v, got %v", DefaultReadRetryAttempts, storage.ReadRetryAttempts)
+	}
 }
 
 func assertDefaultBackupStorage(t *testing.T, storage Storage) {

config/storage.go

@@ -36,6 +36,9 @@ type Storage struct {
 	// copy behaviour control option
 	StorageOptimizeCopy bool `json:"storage_optimize_copy" toml:"storage_optimize_copy" yaml:"storage_optimize_copy"`
 
+	ReadRetryMode     string `json:"read_retry_mode" toml:"read_retry_mode" yaml:"read_retry_mode"`
+	ReadRetryAttempts int    `json:"read_retry_attempts" toml:"read_retry_attempts" yaml:"read_retry_attempts"`
+
 	// File storage default s3. Available: s3, fs
 	StorageType string `json:"storage_type" toml:"storage_type" yaml:"storage_type"`
 
@@ -52,7 +55,9 @@ const (
 	DefaultEndpointSourceScheme = "https"
 
 	/* 1 GB per  second */
-	DefaultStorageRateLimit = 1024 * 1024 * 1024
+	DefaultStorageRateLimit  = 1024 * 1024 * 1024
+	DefaultReadRetryMode     = "resume"
+	DefaultReadRetryAttempts = 3
 )
 
 type StorageOption func(*Storage)
@@ -63,6 +68,11 @@ func WithStorageType(storageType string) StorageOption {
 	}
 }
 
+func WithReadRetryMode(mode string) StorageOption { return func(s *Storage) { s.ReadRetryMode = mode } }
+func WithReadRetryAttempts(attempts int) StorageOption {
+	return func(s *Storage) { s.ReadRetryAttempts = attempts }
+}
+
 func WithStorageConcurrency(storageConcurrency int64) StorageOption {
 	return func(s *Storage) {
 		s.StorageConcurrency = storageConcurrency
@@ -95,6 +105,8 @@ func BuildStorage(opts ...StorageOption) *Storage {
 		WithStorageConcurrency(DefaultStorageConcurrency),
 		WithCopyStorageConcurrency(DefaultCopyStorageConcurrency),
 		WithStorageRateLimit(DefaultStorageRateLimit),
+		WithReadRetryMode(DefaultReadRetryMode),
+		WithReadRetryAttempts(DefaultReadRetryAttempts),
 		WithEndpointSourceScheme(DefaultEndpointSourceScheme),
 	)
 	ApplyStorageOptions(s, opts...)

pkg/proc/interaction.go

@@ -42,6 +42,33 @@ func (*ProtoMgrImpl) ProcessCatExtended(
 	ycl client.YproxyClient) error {
 
 	ycl.SetExternalFilePath(name)
+	if decrypt && cr == nil {
+		err := fmt.Errorf("failed to decrypt object, decrypter not configured")
+		ylogger.Zero.Error().Err(err).Msg("cat failed")
+		return err
+	}
+	if config.InstanceConfig().StorageCnf.ReadRetryMode == "restart_from_beginning" {
+		attempts := config.InstanceConfig().StorageCnf.ReadRetryAttempts
+		if attempts <= 0 {
+			attempts = config.DefaultReadRetryAttempts
+		}
+		open := func() (io.Reader, io.Closer, error) {
+			rawReader := yio.NewYproxyErrorReader(
+				yio.NewRestartReader(s, name, settings), ycl)
+			contentReader := io.Reader(rawReader)
+			if decrypt {
+				var err error
+				contentReader, err = cr.Decrypt(rawReader)
+				if err != nil {
+					_ = rawReader.Close()
+					return nil, nil, err
+				}
+			}
+			return contentReader, rawReader, nil
+		}
+		_, err := copyWithFullRestart(ycl.GetRW(), open, int64(startOffset), attempts)
+		return err
+	}
 
 	yr := yio.NewYRetryReader(yio.NewRestartReader(s, name, settings), ycl)
 
@@ -51,12 +78,6 @@ func (*ProtoMgrImpl) ProcessCatExtended(
 	var err error
 
 	if decrypt {
-		if cr == nil {
-			err := fmt.Errorf("failed to decrypt object, decrypter not configured")
-
-			ylogger.Zero.Error().Err(err).Msg("cat failed")
-			return err
-		}
 		ylogger.Zero.Debug().Str("object-path", name).Msg("decrypt object")
 		contentReader, err = cr.Decrypt(yr)
 		if err != nil {

pkg/proc/retry_copy.go

@@ -0,0 +1,31 @@
+package proc
+
+import (
+	"fmt"
+	"io"
+)
+
+type contentReaderFactory func() (io.Reader, io.Closer, error)
+
+func copyWithFullRestart(dst io.Writer, open contentReaderFactory, skip int64,
+	attempts int) (int64, error) {
+	var delivered int64
+	for attempt := 0; attempt < attempts; attempt++ {
+		reader, closer, err := open()
+		if err != nil {
+			return delivered, err
+		}
+
+		_, err = io.CopyN(io.Discard, reader, skip+delivered)
+		if err == nil {
+			var copied int64
+			copied, err = io.Copy(dst, reader)
+			delivered += copied
+		}
+		closeErr := closer.Close()
+		if err == nil {
+			return delivered, closeErr
+		}
+	}
+	return delivered, fmt.Errorf("failed to read after %d full restart attempts", attempts)
+}

pkg/proc/retry_copy_test.go

@@ -0,0 +1,46 @@
+package proc
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"strings"
+	"testing"
+)
+
+type failingReader struct {
+	reader io.Reader
+	err    error
+}
+
+type testCloser struct{}
+
+func (testCloser) Close() error { return nil }
+
+func (r *failingReader) Read(p []byte) (int, error) {
+	n, err := r.reader.Read(p)
+	if err == io.EOF {
+		return n, r.err
+	}
+	return n, err
+}
+
+func TestCopyWithFullRestartReplaysDeliveredBytes(t *testing.T) {
+	var opens int
+	open := func() (io.Reader, io.Closer, error) {
+		opens++
+		if opens == 1 {
+			return &failingReader{reader: strings.NewReader("abcdef"), err: errors.New("unexpected EOF")}, testCloser{}, nil
+		}
+		return strings.NewReader("abcdef"), testCloser{}, nil
+	}
+
+	var dst bytes.Buffer
+	n, err := copyWithFullRestart(&dst, open, 0, 2)
+	if err != nil {
+		t.Fatalf("copyWithFullRestart failed: %v", err)
+	}
+	if n != 6 || dst.String() != "abcdef" || opens != 2 {
+		t.Fatalf("got n=%d data=%q opens=%d", n, dst.String(), opens)
+	}
+}

pkg/proc/yio/yrreader.go

@@ -90,9 +90,10 @@ type YproxyRetryReader struct {
 
 	selfCl client.YproxyClient
 
-	offsetReached int64
-	retryLimit    int
-	needReacquire bool
+	offsetReached   int64
+	retryLimit      int
+	needReacquire   bool
+	returnReadError bool
 }
 
 // Close implements io.ReadCloser.
@@ -115,6 +116,9 @@ func (y *YproxyRetryReader) Read(p []byte) (int, error) {
 			err := y.underlying.Restart(y.offsetReached)
 
 			if err != nil {
+				if y.returnReadError {
+					return 0, err
+				}
 				// log error and continue.
 				// Try to mitigate overload problems with random sleep
 				ylogger.Zero.Error().Err(err).Int("offset reached", int(y.offsetReached)).Int("retry count", int(retry)).Msg("failed to reacquire external storage connection, wait and retry")
@@ -143,6 +147,7 @@ func (y *YproxyRetryReader) Read(p []byte) (int, error) {
 
 			if n > 0 {
 				y.offsetReached += int64(n)
+				y.selfCl.SetByteOffset(y.offsetReached)
 			}
 
 			// what if close failed?
@@ -152,6 +157,12 @@ func (y *YproxyRetryReader) Read(p []byte) (int, error) {
 			// from previously reached point
 
 			y.needReacquire = true
+			if y.returnReadError {
+				return n, err
+			}
+			if n > 0 {
+				return n, nil
+			}
 			continue
 		} else {
 			y.offsetReached += int64(n)
@@ -167,6 +178,13 @@ func (y *YproxyRetryReader) Read(p []byte) (int, error) {
 	return -1, fmt.Errorf("failed to read within retries: no error captured")
 }
 
+// NewYproxyErrorReader returns storage errors to the caller instead of resuming
+// from a Range offset. It is used when the whole content stream is retried.
+func NewYproxyErrorReader(r RestartReader, selfCl client.YproxyClient) io.ReadCloser {
+	return &YproxyRetryReader{underlying: r, retryLimit: 1, selfCl: selfCl,
+		needReacquire: true, returnReadError: true}
+}
+
 const (
 	defaultRetryLimit = 100
 )

pkg/proc/yio/yrreader_test.go

@@ -135,3 +135,38 @@ func TestYproxyRetryReaderSimpleReadRetry(t *testing.T) {
 
 	assert.Nil(t, yr.Close())
 }
+
+func TestYproxyRetryReaderReturnsPartialReadBeforeRetry(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	rr := mockio.NewMockRestartReader(ctrl)
+	ycl := mockcl.NewMockYproxyClient(ctrl)
+	yr := yio.NewYRetryReader(rr, ycl)
+
+	rr.EXPECT().Restart(int64(0)).Return(nil)
+	rr.EXPECT().Read(gomock.Any()).Do(func(buf []byte) {
+		copy(buf, "ab")
+	}).Return(2, fmt.Errorf("unexpected EOF"))
+	rr.EXPECT().Close().Return(nil)
+	ycl.EXPECT().SetByteOffset(int64(2))
+
+	first := make([]byte, 2)
+	n, err := yr.Read(first)
+	assert.NoError(t, err)
+	assert.Equal(t, 2, n)
+	assert.Equal(t, []byte("ab"), first)
+
+	rr.EXPECT().Restart(int64(2)).Return(nil)
+	rr.EXPECT().Read(gomock.Any()).Do(func(buf []byte) {
+		copy(buf, []byte{3, 4, 5})
+	}).Return(3, nil)
+	ycl.EXPECT().SetByteOffset(int64(5))
+
+	second := make([]byte, 3)
+	n, err = yr.Read(second)
+	assert.NoError(t, err)
+	assert.Equal(t, 3, n)
+	assert.Equal(t, []byte{3, 4, 5}, second)
+
+	rr.EXPECT().Close().Return(nil)
+	assert.NoError(t, yr.Close())
+}