From 996926dfe36a8c1846f2317bef2633c1044e538b Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Mon, 26 Oct 2026 04:29:40 +0000 Subject: [PATCH 1/3] fix: preserve local stream reader offsets Signed-off-by: Perfloop Agent --- docs/embedded.md | 6 +- embedded/client.go | 6 +- embedded/put_stream_test.go | 120 ++++++++++++++++++++++ server/operations/put.go | 44 +++++--- server/operations/put_local_retry_test.go | 112 ++++++++++++++++++++ 5 files changed, 273 insertions(+), 15 deletions(-) create mode 100644 embedded/put_stream_test.go create mode 100644 server/operations/put_local_retry_test.go diff --git a/docs/embedded.md b/docs/embedded.md index 05c438c8..688ea455 100644 --- a/docs/embedded.md +++ b/docs/embedded.md @@ -269,7 +269,11 @@ keys, nextToken, hasMore, err = cache.ListPage(ctx, prefix, limit, continuationT ### Streaming Operations -For large values, use streaming to avoid loading everything into memory: +For large values, use streaming to avoid loading everything into memory. +`PutStream` reads from the reader's current position through EOF. A local +retryable failure restarts from that call-entry position when it can be obtained +through `io.Seeker`; otherwise the local write is attempted once. A failed +retry seek returns an error. ```go // Stream data to cache diff --git a/embedded/client.go b/embedded/client.go index c5598572..ff4fb482 100644 --- a/embedded/client.go +++ b/embedded/client.go @@ -417,7 +417,11 @@ func (c *Client) ListPageWithValues(ctx context.Context, prefix string, limit in return entries, token, more, nil } -// PutStream stores data from a reader for the given key. +// PutStream stores bytes from r's current position through EOF. On local writes, +// the first attempt does not rewind r; retryable failures restart from the +// call-entry position when it can be obtained through io.Seeker. Readers with +// no obtainable position are attempted once. A failed retry seek returns an +// error. func (c *Client) PutStream(ctx context.Context, key string, r io.Reader, ttlSeconds int64) error { return c.ops.Put(ctx, key, r, int(ttlSeconds)) } diff --git a/embedded/put_stream_test.go b/embedded/put_stream_test.go new file mode 100644 index 00000000..4e1e2946 --- /dev/null +++ b/embedded/put_stream_test.go @@ -0,0 +1,120 @@ +// Copyright 2026 Tigris Data, Inc. +// SPDX-License-Identifier: Apache-2.0 + +package embedded + +import ( + "bytes" + "context" + "io" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" + stor "github.com/tigrisdata/ocache/storage" +) + +const putStreamReaderPositionInlineThreshold = 8 + +// readerOnly hides the seek methods of its underlying reader. +type readerOnly struct { + io.Reader +} + +func TestPutStreamReaderPosition(t *testing.T) { + client, err := New(&Config{ + DiskPath: t.TempDir(), + TTL: time.Hour, + InlineThreshold: putStreamReaderPositionInlineThreshold, + Storage: &stor.StorageConfig{ + CleanupInterval: time.Hour, + DisableRecompaction: true, + }, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = client.Close() }) + + ctx := context.Background() + storeAndCheck := func(t *testing.T, key string, body io.Reader, want []byte) { + t.Helper() + require.NoError(t, client.PutStream(ctx, key, body, 0)) + got, err := client.Get(ctx, key) + require.NoError(t, err) + require.NotNil(t, got, "PutStream should create the key even when the unread suffix is empty") + require.Equal(t, want, got) + } + + t.Run("bytes reader starts at current position", func(t *testing.T) { + data := []byte("consumed-prefix|unread-bytes") + position := int64(len("consumed-prefix|")) + body := bytes.NewReader(data) + _, err := body.Seek(position, io.SeekStart) + require.NoError(t, err) + storeAndCheck(t, "bytes-position", body, data[position:]) + }) + + t.Run("file starts at current position", func(t *testing.T) { + data := []byte("file-prefix|remaining-file-data") + position := int64(len("file-prefix|")) + path := filepath.Join(t.TempDir(), "body") + require.NoError(t, os.WriteFile(path, data, 0o600)) + body, err := os.Open(path) + require.NoError(t, err) + t.Cleanup(func() { _ = body.Close() }) + _, err = body.Seek(position, io.SeekStart) + require.NoError(t, err) + storeAndCheck(t, "file-position", body, data[position:]) + }) + + t.Run("zero offset", func(t *testing.T) { + data := []byte("whole value from zero") + storeAndCheck(t, "zero-position", bytes.NewReader(data), data) + }) + + t.Run("end of input", func(t *testing.T) { + body := bytes.NewReader([]byte("already consumed")) + _, err := body.Seek(0, io.SeekEnd) + require.NoError(t, err) + storeAndCheck(t, "end-position", body, []byte{}) + }) + + t.Run("non-seekable reader starts at current position", func(t *testing.T) { + data := []byte("wrapped-prefix|non-seekable-remainder") + position := int64(len("wrapped-prefix|")) + body := bytes.NewReader(data) + _, err := body.Seek(position, io.SeekStart) + require.NoError(t, err) + storeAndCheck(t, "non-seekable-position", readerOnly{Reader: body}, data[position:]) + }) + + t.Run("pipe-backed file is readable when seek is unsupported", func(t *testing.T) { + data := []byte("pipe data from the reader's current position") + readEnd, writeEnd, err := os.Pipe() + require.NoError(t, err) + writerDone := make(chan error, 1) + go func() { + n, writeErr := writeEnd.Write(data) + closeErr := writeEnd.Close() + if writeErr == nil && n != len(data) { + writeErr = io.ErrShortWrite + } + if writeErr != nil { + writerDone <- writeErr + return + } + writerDone <- closeErr + }() + + putErr := client.PutStream(ctx, "pipe-file", readEnd, 0) + readCloseErr := readEnd.Close() + writeErr := <-writerDone + require.NoError(t, putErr) + require.NoError(t, readCloseErr) + require.NoError(t, writeErr) + got, err := client.Get(ctx, "pipe-file") + require.NoError(t, err) + require.Equal(t, data, got) + }) +} diff --git a/server/operations/put.go b/server/operations/put.go index 8768b3d0..f14a0b86 100644 --- a/server/operations/put.go +++ b/server/operations/put.go @@ -73,10 +73,10 @@ func (o *Operations) OverBudget() bool { return o.storage.OverShedRatio() } -// PutLocal stores data in local storage directly. -// If the body implements io.Seeker (e.g., bytes.Reader), retries are enabled with automatic reset. -// For non-seekable readers (e.g., io.PipeReader from streaming), no retries are attempted -// since the reader cannot be rewound. +// PutLocal stores data in local storage directly. If the reader's current +// position can be obtained through io.Seeker, retryable failures restart from +// that call-entry position. Otherwise, storage is attempted once from the +// reader's current position. func (o *Operations) PutLocal(ctx context.Context, key string, body io.Reader, ttl int) error { return o.PutLocalWithOptions(ctx, key, body, ttl, PutOptions{}) } @@ -88,20 +88,38 @@ func (o *Operations) PutLocalWithOptions(ctx context.Context, key string, body i return storageErrors.NewOverBudgetError("Put") } - // Check if we can retry (reader is seekable) - seeker, canRetry := body.(io.Seeker) - - if !canRetry { - // Non-seekable reader (e.g., io.PipeReader) - no retry possible + return putLocalWithRetry(ctx, key, body, func() error { return o.storage.Put(key, body, ttl) + }) +} + +// putLocalWithRetry attempts an unpositionable reader once. When its initial +// offset is available, the first attempt reads from the current position and +// each retry restores that same offset. +func putLocalWithRetry(ctx context.Context, key string, body io.Reader, put func() error) error { + seeker, ok := body.(io.Seeker) + if !ok { + return put() } - // Seekable reader (e.g., bytes.Reader) - retry with reset + start, err := seeker.Seek(0, io.SeekCurrent) + if err != nil || start < 0 { + return put() + } + + attempt := 0 return retry.DoWithKey(ctx, retry.DefaultConfig(), "Put", key, func() error { - if _, err := seeker.Seek(0, io.SeekStart); err != nil { - return err + if attempt > 0 { + if _, err := seeker.Seek(start, io.SeekStart); err != nil { + if storageErrors.IsRetryable(err) { + // A failed reposition is terminal even if its error resembles a storage error. + return errors.New(err.Error()) + } + return err + } } - return o.storage.Put(key, body, ttl) + attempt++ + return put() }) } diff --git a/server/operations/put_local_retry_test.go b/server/operations/put_local_retry_test.go new file mode 100644 index 00000000..d3e137bd --- /dev/null +++ b/server/operations/put_local_retry_test.go @@ -0,0 +1,112 @@ +// Copyright 2026 Tigris Data, Inc. +// SPDX-License-Identifier: Apache-2.0 + +package operations + +import ( + "bytes" + "context" + "errors" + "io" + "testing" + + "github.com/stretchr/testify/require" + storageErrors "github.com/tigrisdata/ocache/storage/errors" +) + +var errRetryPosition = errors.New("injected retry reposition failure") + +type failRetrySeekReader struct { + *bytes.Reader + start int64 + consumed bool + restoreErr error +} + +func (r *failRetrySeekReader) Read(p []byte) (int, error) { + n, err := r.Reader.Read(p) + if n > 0 { + r.consumed = true + } + return n, err +} + +func (r *failRetrySeekReader) Seek(offset int64, whence int) (int64, error) { + if r.consumed && whence == io.SeekStart && offset == r.start { + return 0, r.restoreErr + } + return r.Reader.Seek(offset, whence) +} + +func TestPutLocalReaderRetryPosition(t *testing.T) { + data := []byte("already-consumed|replay-this-suffix") + start := int64(len("already-consumed|")) + body := bytes.NewReader(data) + _, err := body.Seek(start, io.SeekStart) + require.NoError(t, err) + + var attempts int + var firstAttempt, retryAttempt []byte + put := func() error { + attempts++ + if attempts == 1 { + partial := make([]byte, 5) + n, err := body.Read(partial) + require.NoError(t, err) + firstAttempt = append([]byte(nil), partial[:n]...) + return storageErrors.NewTemporaryError("Put", "key", errors.New("injected storage failure")) + } + var err error + retryAttempt, err = io.ReadAll(body) + return err + } + + require.NoError(t, putLocalWithRetry(context.Background(), "key", body, put)) + require.Equal(t, 2, attempts) + require.Equal(t, data[start:start+5], firstAttempt, "the initial attempt must read from the call-entry position") + require.Equal(t, data[start:], retryAttempt, "the retry must replay the full unread suffix") +} + +func TestPutLocalReaderRetrySeekFailure(t *testing.T) { + for _, tc := range []struct { + name string + seekError error + }{ + {name: "ordinary seek error", seekError: errRetryPosition}, + {name: "retryable-typed seek error", seekError: storageErrors.NewTemporaryError("Seek", "key", errors.New("injected seek failure"))}, + } { + t.Run(tc.name, func(t *testing.T) { + data := []byte("prefix|body-that-fails-before-commit") + start := int64(len("prefix|")) + body := &failRetrySeekReader{ + Reader: bytes.NewReader(data), + start: start, + restoreErr: tc.seekError, + } + _, err := body.Seek(start, io.SeekStart) + require.NoError(t, err) + + var attempts int + put := func() error { + attempts++ + if attempts == 1 { + partial := make([]byte, 5) + _, err := body.Read(partial) + require.NoError(t, err) + return storageErrors.NewTemporaryError("Put", "key", errors.New("injected storage failure")) + } + return nil + } + + err = putLocalWithRetry(context.Background(), "key", body, put) + require.EqualError(t, err, tc.seekError.Error()) + if tc.seekError == errRetryPosition { + require.ErrorIs(t, err, errRetryPosition) + } else { + require.False(t, storageErrors.IsRetryable(err), "a failed reposition must stop the storage retry loop") + } + require.True(t, body.consumed) + require.Equal(t, 1, attempts, "storage must not be called after a failed retry reposition") + }) + } +} From 578636eaa8d9090fa0cdb74ce243cc195c275b82 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Mon, 26 Oct 2026 05:41:03 +0000 Subject: [PATCH 2/3] fix: preserve cancellation for unpositionable streams Signed-off-by: Perfloop Agent --- embedded/put_stream_test.go | 28 ++++++++++++++++ server/operations/put.go | 6 +++- server/operations/put_local_retry_test.go | 40 +++++++++++++++++++++++ 3 files changed, 73 insertions(+), 1 deletion(-) diff --git a/embedded/put_stream_test.go b/embedded/put_stream_test.go index 4e1e2946..d020d5fd 100644 --- a/embedded/put_stream_test.go +++ b/embedded/put_stream_test.go @@ -14,6 +14,7 @@ import ( "github.com/stretchr/testify/require" stor "github.com/tigrisdata/ocache/storage" + storageErrors "github.com/tigrisdata/ocache/storage/errors" ) const putStreamReaderPositionInlineThreshold = 8 @@ -117,4 +118,31 @@ func TestPutStreamReaderPosition(t *testing.T) { require.NoError(t, err) require.Equal(t, data, got) }) + + t.Run("canceled request does not consume or store an unpositionable seeker", func(t *testing.T) { + data := []byte("already-canceled pipe payload") + readEnd, writeEnd, err := os.Pipe() + require.NoError(t, err) + t.Cleanup(func() { _ = readEnd.Close() }) + t.Cleanup(func() { _ = writeEnd.Close() }) + + n, err := writeEnd.Write(data) + require.NoError(t, err) + require.Equal(t, len(data), n) + require.NoError(t, writeEnd.Close()) + + canceled, cancel := context.WithCancel(ctx) + cancel() + putErr := client.PutStream(canceled, "canceled-pipe", readEnd, 0) + gotType, ok := storageErrors.GetType(putErr) + require.True(t, ok, "a canceled seekable input must retain its timeout error") + require.Equal(t, storageErrors.TypeTimeout, gotType) + + remaining, err := io.ReadAll(readEnd) + require.NoError(t, err) + require.Equal(t, data, remaining, "cancellation before the storage attempt must not consume the reader") + got, err := client.Get(ctx, "canceled-pipe") + require.NoError(t, err) + require.Nil(t, got, "cancellation before the storage attempt must not create the key") + }) } diff --git a/server/operations/put.go b/server/operations/put.go index f14a0b86..a0d9763b 100644 --- a/server/operations/put.go +++ b/server/operations/put.go @@ -104,7 +104,11 @@ func putLocalWithRetry(ctx context.Context, key string, body io.Reader, put func start, err := seeker.Seek(0, io.SeekCurrent) if err != nil || start < 0 { - return put() + // Keep DoWithKey's pre-attempt cancellation check, but do not retry a + // reader whose original position cannot be restored. + cfg := retry.DefaultConfig() + cfg.MaxRetries = 0 + return retry.DoWithKey(ctx, cfg, "Put", key, put) } attempt := 0 diff --git a/server/operations/put_local_retry_test.go b/server/operations/put_local_retry_test.go index d3e137bd..195051ed 100644 --- a/server/operations/put_local_retry_test.go +++ b/server/operations/put_local_retry_test.go @@ -16,6 +16,17 @@ import ( var errRetryPosition = errors.New("injected retry reposition failure") +type unavailablePositionReader struct { + *bytes.Reader +} + +func (r *unavailablePositionReader) Seek(offset int64, whence int) (int64, error) { + if whence == io.SeekCurrent && offset == 0 { + return 0, errRetryPosition + } + return r.Reader.Seek(offset, whence) +} + type failRetrySeekReader struct { *bytes.Reader start int64 @@ -67,6 +78,35 @@ func TestPutLocalReaderRetryPosition(t *testing.T) { require.Equal(t, data[start:], retryAttempt, "the retry must replay the full unread suffix") } +func TestPutLocalReaderWithoutPositionAttemptsOnce(t *testing.T) { + data := []byte("already-consumed|read-once-suffix") + start := int64(len("already-consumed|")) + reader := bytes.NewReader(data) + _, err := reader.Seek(start, io.SeekStart) + require.NoError(t, err) + body := &unavailablePositionReader{Reader: reader} + + var attempts int + var firstRead []byte + storageFailure := errors.New("injected storage failure") + put := func() error { + attempts++ + buf := make([]byte, 5) + n, err := body.Read(buf) + require.NoError(t, err) + if attempts == 1 { + firstRead = append([]byte(nil), buf[:n]...) + return storageErrors.NewTemporaryError("Put", "key", storageFailure) + } + return nil + } + + err = putLocalWithRetry(context.Background(), "key", body, put) + require.ErrorIs(t, err, storageFailure) + require.Equal(t, 1, attempts, "an unavailable starting position must disable retries") + require.Equal(t, data[start:start+5], firstRead, "the single attempt must read from the current position") +} + func TestPutLocalReaderRetrySeekFailure(t *testing.T) { for _, tc := range []struct { name string From c9fed336c58bcd8ce2dd6181a1598812d9ec8ab6 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Mon, 26 Oct 2026 07:06:52 +0000 Subject: [PATCH 3/3] fix: check cancellation before reader positioning Signed-off-by: Perfloop Agent --- embedded/put_stream_test.go | 55 +++++++++++++++++++++++++++++++++++++ server/operations/put.go | 4 +++ 2 files changed, 59 insertions(+) diff --git a/embedded/put_stream_test.go b/embedded/put_stream_test.go index d020d5fd..b3b61af2 100644 --- a/embedded/put_stream_test.go +++ b/embedded/put_stream_test.go @@ -24,6 +24,23 @@ type readerOnly struct { io.Reader } +type blockingPositionSeeker struct { + *bytes.Reader + seekEntered chan struct{} + releaseSeek <-chan struct{} +} + +func (r *blockingPositionSeeker) Seek(offset int64, whence int) (int64, error) { + if offset == 0 && whence == io.SeekCurrent { + select { + case r.seekEntered <- struct{}{}: + default: + } + <-r.releaseSeek + } + return r.Reader.Seek(offset, whence) +} + func TestPutStreamReaderPosition(t *testing.T) { client, err := New(&Config{ DiskPath: t.TempDir(), @@ -145,4 +162,42 @@ func TestPutStreamReaderPosition(t *testing.T) { require.NoError(t, err) require.Nil(t, got, "cancellation before the storage attempt must not create the key") }) + + t.Run("pre-canceled request returns before a blocking position query", func(t *testing.T) { + data := []byte("already-canceled blocking-seeker payload") + releaseSeek := make(chan struct{}) + body := &blockingPositionSeeker{ + Reader: bytes.NewReader(data), + seekEntered: make(chan struct{}, 1), + releaseSeek: releaseSeek, + } + canceled, cancel := context.WithCancel(ctx) + cancel() + done := make(chan error, 1) + go func() { + done <- client.PutStream(canceled, "canceled-blocking-seeker", body, 0) + }() + + select { + case putErr := <-done: + gotType, ok := storageErrors.GetType(putErr) + require.True(t, ok, "a canceled seekable input must retain its timeout error") + require.Equal(t, storageErrors.TypeTimeout, gotType) + case <-body.seekEntered: + close(releaseSeek) + putErr := <-done + t.Fatalf("PutStream queried the reader before the canceled-context check: %v", putErr) + case <-time.After(5 * time.Second): + close(releaseSeek) + putErr := <-done + t.Fatalf("PutStream did not return before a blocked position query: %v", putErr) + } + + remaining, err := io.ReadAll(body.Reader) + require.NoError(t, err) + require.Equal(t, data, remaining, "the canceled request must not consume the reader") + got, err := client.Get(ctx, "canceled-blocking-seeker") + require.NoError(t, err) + require.Nil(t, got, "the canceled request must not create the key") + }) } diff --git a/server/operations/put.go b/server/operations/put.go index a0d9763b..9adefc78 100644 --- a/server/operations/put.go +++ b/server/operations/put.go @@ -101,6 +101,10 @@ func putLocalWithRetry(ctx context.Context, key string, body io.Reader, put func if !ok { return put() } + if ctx.Err() != nil { + // Match DoWithKey's pre-attempt check before calling a caller-supplied Seek. + return storageErrors.NewTimeoutError("Put", key) + } start, err := seeker.Seek(0, io.SeekCurrent) if err != nil || start < 0 {