objstorage: check for missing file in remoteReadable

An external file can go missing after we "open" it (i.e. after we
obtain a `remote.ObjectReader`). This should be marked as a corruption
error; this commit fixes this and adds tests.

Author: RaduBerinde

objstorage/objstorageprovider/remote.go

@@ -348,7 +348,7 @@ func (p *provider) remoteOpenForReading(
 		}
 		return nil, err
 	}
-	return p.newRemoteReadable(reader, size, meta.DiskFileNum), nil
+	return p.newRemoteReadable(reader, size, meta.DiskFileNum, meta.Remote.Storage.IsNotExistError), nil
 }
 
 func (p *provider) remoteSize(meta objstorage.ObjectMetadata) (int64, error) {

objstorage/objstorageprovider/remote_backing.go

@@ -263,7 +263,7 @@ func (p *provider) AttachRemoteObjects(
 			_ = p.sharedUnref(d.meta)
 			// TODO(radu): clean up references previously created in this loop.
 			if d.meta.Remote.Storage.IsNotExistError(err) {
-				return nil, errors.Errorf("origin marker object %q does not exist;"+
+				return nil, base.CorruptionErrorf("origin marker object %q does not exist;"+
 					" object probably removed from the provider which created the backing", refName)
 			}
 			return nil, errors.Wrapf(err, "checking origin's marker object %s", refName)

objstorage/objstorageprovider/remote_readable.go

@@ -35,22 +35,27 @@ const remoteReadaheadSizeForCompaction = 8 * 1024 * 1024 /* 8MB */
 // remote.ObjectReader returned by remote.Storage.ReadObject. It is stateless
 // and can be called concurrently.
 type remoteReadable struct {
-	objReader remote.ObjectReader
-	size      int64
-	fileNum   base.DiskFileNum
-	cache     *sharedcache.Cache
+	objReader     remote.ObjectReader
+	size          int64
+	fileNum       base.DiskFileNum
+	cache         *sharedcache.Cache
+	errIsNotExist func(error) bool
 }
 
 var _ objstorage.Readable = (*remoteReadable)(nil)
 
 func (p *provider) newRemoteReadable(
-	objReader remote.ObjectReader, size int64, fileNum base.DiskFileNum,
+	objReader remote.ObjectReader,
+	size int64,
+	fileNum base.DiskFileNum,
+	errIsNotExist func(error) bool,
 ) *remoteReadable {
 	return &remoteReadable{
-		objReader: objReader,
-		size:      size,
-		fileNum:   fileNum,
-		cache:     p.remote.cache,
+		objReader:     objReader,
+		size:          size,
+		fileNum:       fileNum,
+		cache:         p.remote.cache,
+		errIsNotExist: errIsNotExist,
 	}
 }
 
@@ -64,14 +69,21 @@ func (r *remoteReadable) ReadAt(ctx context.Context, p []byte, offset int64) err
 func (r *remoteReadable) readInternal(
 	ctx context.Context, p []byte, offset int64, forCompaction bool,
 ) error {
+	var err error
 	if r.cache != nil {
 		flags := sharedcache.ReadFlags{
 			// Don't add data to the cache if this read is for a compaction.
 			ReadOnly: forCompaction,
 		}
-		return r.cache.ReadAt(ctx, r.fileNum, p, offset, r.objReader, r.size, flags)
+		err = r.cache.ReadAt(ctx, r.fileNum, p, offset, r.objReader, r.size, flags)
+	} else {
+		err = r.objReader.ReadAt(ctx, p, offset)
+	}
+	if err != nil && r.errIsNotExist(err) {
+		// If a file goes missing, we consider this a corruption error.
+		err = base.MarkCorruptionError(err)
 	}
-	return r.objReader.ReadAt(ctx, p, offset)
+	return err
 }
 
 func (r *remoteReadable) Close() error {

objstorage/objstorageprovider/remote_readable_test.go

@@ -13,7 +13,10 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/datadriven"
+	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/objstorage"
+	"github.com/cockroachdb/pebble/objstorage/remote"
+	"github.com/cockroachdb/pebble/vfs"
 	"github.com/stretchr/testify/require"
 )
 
@@ -106,3 +109,40 @@ func TestRemoteReadHandle(t *testing.T) {
 		}
 	})
 }
+
+// TestErrorWhenObjectDisappears verifies that the provider returns a corruption
+// error when we read from an opened object that disappears from under us.
+func TestErrorWhenObjectDisappears(t *testing.T) {
+	remoteStorage := remote.NewInMem()
+	settings := DefaultSettings(vfs.NewMem(), "")
+	settings.Remote.StorageFactory = remote.MakeSimpleFactory(map[remote.Locator]remote.Storage{
+		"locator": remoteStorage,
+	})
+	settings.Remote.CreateOnSharedLocator = "locator"
+	settings.Remote.CreateOnShared = remote.CreateOnSharedAll
+	provider, err := Open(settings)
+	require.NoError(t, err)
+	defer provider.Close()
+	require.NoError(t, provider.SetCreatorID(1))
+
+	ctx := context.Background()
+	writable, objMeta, err := provider.Create(ctx, base.FileTypeTable, 1, objstorage.CreateOptions{
+		PreferSharedStorage: true,
+	})
+	require.NoError(t, err)
+	require.NotNil(t, objMeta.Remote.Storage)
+	require.NoError(t, writable.Write([]byte("hello")))
+	require.NoError(t, writable.Finish())
+
+	readable, err := provider.OpenForReading(ctx, base.FileTypeTable, 1, objstorage.OpenOptions{})
+	require.NoError(t, err)
+
+	// Delete all objects from the store and expect to get a corruption error.
+	objects, err := remoteStorage.List("", "")
+	require.NoError(t, err)
+	for _, o := range objects {
+		require.NoError(t, remoteStorage.Delete(o))
+	}
+	err = readable.ReadAt(ctx, make([]byte, 1), 0)
+	require.True(t, base.IsCorruptionError(err))
+}

objstorage/objstorageprovider/testdata/provider/shared_attach_after_unref

@@ -83,7 +83,7 @@ p5b2 102
 ----
 <remote> create object "1ab5-5-000002.sst.ref.6.000102"
 <remote> close writer for "1ab5-5-000002.sst.ref.6.000102" after 0 bytes
-<remote> size of object "1ab5-5-000002.sst.ref.5.000002": error: file does not exist
+<remote> size of object "1ab5-5-000002.sst.ref.5.000002": error: in-mem remote storage object does not exist
 <remote> delete object "1ab5-5-000002.sst.ref.6.000102"
 <remote> list (prefix="1ab5-5-000002.sst.ref.", delimiter="")
 <remote> delete object "1ab5-5-000002.sst"

objstorage/remote/mem.go

@@ -8,9 +8,10 @@ import (
 	"bytes"
 	"context"
 	"io"
-	"os"
 	"strings"
 	"sync"
+
+	"github.com/cockroachdb/errors"
 )
 
 // NewInMem returns an in-memory implementation of the remote.Storage
@@ -49,25 +50,32 @@ func (s *inMemStore) ReadObject(
 	if err != nil {
 		return nil, 0, err
 	}
-	return &inMemReader{data: obj.data}, int64(len(obj.data)), nil
+	return &inMemReader{objName: objName, store: s}, int64(len(obj.data)), nil
 }
 
 type inMemReader struct {
-	data []byte
+	objName string
+	store   *inMemStore
 }
 
 var _ ObjectReader = (*inMemReader)(nil)
 
 func (r *inMemReader) ReadAt(ctx context.Context, p []byte, offset int64) error {
-	if offset+int64(len(p)) > int64(len(r.data)) {
+	// We don't just store obj.data in the inMemReader because we want to emit an
+	// error if the object is deleted from under us.
+	obj, err := r.store.getObj(r.objName)
+	if err != nil {
+		return err
+	}
+	if offset+int64(len(p)) > int64(len(obj.data)) {
 		return io.EOF
 	}
-	copy(p, r.data[offset:])
+	copy(p, obj.data[offset:])
 	return nil
 }
 
 func (r *inMemReader) Close() error {
-	r.data = nil
+	r.store = nil
 	return nil
 }
 
@@ -135,15 +143,19 @@ func (s *inMemStore) Size(objName string) (int64, error) {
 }
 
 func (s *inMemStore) IsNotExistError(err error) bool {
-	return err == os.ErrNotExist
+	return errors.Is(err, inMemStoreNotExistErr)
 }
 
+// We use a custom "not exists" error to make sure that callers correctly use
+// IsNotExistError.
+var inMemStoreNotExistErr = errors.Newf("in-mem remote storage object does not exist")
+
 func (s *inMemStore) getObj(name string) (*inMemObj, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	obj, ok := s.mu.objects[name]
 	if !ok {
-		return nil, os.ErrNotExist
+		return nil, inMemStoreNotExistErr
 	}
 	return obj, nil
 }

sstable/reader_test.go

@@ -33,6 +33,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/objstorage"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
+	"github.com/cockroachdb/pebble/objstorage/remote"
 	"github.com/cockroachdb/pebble/sstable/block"
 	"github.com/cockroachdb/pebble/sstable/valblk"
 	"github.com/cockroachdb/pebble/vfs"
@@ -2535,3 +2536,68 @@ func newReader(r ReadableFile, o ReaderOptions) (*Reader, error) {
 	}
 	return NewReader(context.Background(), readable, o)
 }
+
+// TestReaderReportsCorruption tests that the reader reports corruption when
+// an external file goes missing after obtaining a remote.ObjectReader for it.
+func TestReaderReportsCorruption(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	ctx := context.Background()
+	remoteStorage := remote.NewInMem()
+	settings := objstorageprovider.DefaultSettings(vfs.NewMem(), "")
+	settings.Remote.StorageFactory = remote.MakeSimpleFactory(map[remote.Locator]remote.Storage{
+		"locator": remoteStorage,
+	})
+	settings.Remote.CreateOnSharedLocator = "locator"
+	settings.Remote.CreateOnShared = remote.CreateOnSharedAll
+	provider, err := objstorageprovider.Open(settings)
+	require.NoError(t, err)
+	defer provider.Close()
+	require.NoError(t, provider.SetCreatorID(1))
+
+	writable, objMeta, err := provider.Create(ctx, base.FileTypeTable, 1, objstorage.CreateOptions{
+		PreferSharedStorage: true,
+	})
+	require.NoError(t, err)
+	require.NotNil(t, objMeta.Remote.Storage)
+
+	// Create an sst file with multiple data blocks.
+	w := NewWriter(writable, WriterOptions{BlockSize: 1})
+	for i := range 100 {
+		require.NoError(t, w.Set([]byte(fmt.Sprintf("%04d", i)), []byte(fmt.Sprintf("value%04d", i))))
+	}
+	require.NoError(t, w.Close())
+
+	readable, err := provider.OpenForReading(ctx, base.FileTypeTable, 1, objstorage.OpenOptions{})
+	require.NoError(t, err)
+	r, err := NewReader(context.Background(), readable, ReaderOptions{})
+	require.NoError(t, err)
+	defer r.Close()
+
+	var lastReportedCorruption error
+	env := block.ReadEnv{
+		ReportCorruptionFn: func(opaque any, err error) error {
+			lastReportedCorruption = err
+			return errors.Wrap(err, "error passed through ReportCorruptionFn")
+		},
+	}
+	iter, err := r.NewPointIter(context.Background(), NoTransforms, nil, nil, nil, 0, env, nil)
+	require.NoError(t, err)
+	defer iter.Close()
+	kv := iter.First()
+	require.NotNil(t, kv)
+
+	// Delete all objects from the store and expect to get a corruption error.
+	objects, err := remoteStorage.List("", "")
+	require.NoError(t, err)
+	for _, o := range objects {
+		require.NoError(t, remoteStorage.Delete(o))
+	}
+	for ; kv != nil; kv = iter.Next() {
+	}
+	iterErr := iter.Error()
+	require.ErrorContains(t, iterErr, "error passed through ReportCorruptionFn: in-mem remote storage object does not exist")
+	require.True(t, base.IsCorruptionError(iterErr))
+	require.ErrorContains(t, lastReportedCorruption, "in-mem remote storage object does not exist")
+	require.True(t, base.IsCorruptionError(lastReportedCorruption))
+}