fix: avoid reporting dedup load sentinel on cache miss

pkg/objectio/cache.go

@@ -91,9 +91,10 @@ var metaLoadMu sync.Mutex
 var metaLoadCalls = make(map[mataCacheKey]*loadCall)
 
 type loadCall struct {
-	done chan struct{}
-	val  []byte
-	err  error
+	done      chan struct{}
+	val       []byte
+	err       error
+	completed bool
 }
 
 func metaCacheSize() int64 {
@@ -345,13 +346,15 @@ func dedupLoad(ctx context.Context, key mataCacheKey, load func() ([]byte, error
 			if call.err != nil {
 				return nil, call.err
 			}
+			if !call.completed {
+				return nil, moerr.NewInternalErrorNoCtx("dedup load did not complete")
+			}
 			return call.val, nil
 		case <-ctx.Done():
 			return nil, ctx.Err()
 		}
 	}
 	call := &loadCall{done: make(chan struct{})}
-	call.err = moerr.NewInternalErrorNoCtx("dedup load did not complete")
 	metaLoadCalls[key] = call
 	metaLoadMu.Unlock()
 
@@ -363,6 +366,7 @@ func dedupLoad(ctx context.Context, key mataCacheKey, load func() ([]byte, error
 	}()
 
 	call.val, call.err = load()
+	call.completed = true
 	if call.err == nil {
 		if target, ok := metaCachePressureTarget(metaCache.Capacity()); ok &&
 			metaCache.Used()+int64(len(call.val)) > target {

pkg/objectio/meta_test.go

@@ -86,6 +86,83 @@ func TestDedupLoadCleansUpAfterPanic(t *testing.T) {
 	assert.Equal(t, []byte("ok"), v)
 }
 
+func TestDedupLoadWaiterGetsSuccessfulOwnerValue(t *testing.T) {
+	oldMetaCache := metaCache
+	metaCache = newMetaCache(fscache.ConstCapacity(1024))
+	defer func() {
+		metaCache = oldMetaCache
+	}()
+
+	var key mataCacheKey
+	key[0] = 8
+	started := make(chan struct{})
+	release := make(chan struct{})
+	waiterDone := make(chan struct {
+		val []byte
+		err error
+	}, 1)
+
+	go func() {
+		_, _ = dedupLoad(context.Background(), key, func() ([]byte, error) {
+			close(started)
+			<-release
+			return []byte("ok"), nil
+		})
+	}()
+	<-started
+
+	go func() {
+		v, err := dedupLoad(context.Background(), key, func() ([]byte, error) {
+			return nil, errors.New("unexpected waiter load")
+		})
+		waiterDone <- struct {
+			val []byte
+			err error
+		}{v, err}
+	}()
+
+	time.Sleep(10 * time.Millisecond)
+	close(release)
+	result := <-waiterDone
+	assert.NoError(t, result.err)
+	assert.Equal(t, []byte("ok"), result.val)
+}
+
+func TestDedupLoadWaiterTimeoutWhileOwnerStillLoading(t *testing.T) {
+	oldMetaCache := metaCache
+	metaCache = newMetaCache(fscache.ConstCapacity(1024))
+	defer func() {
+		metaCache = oldMetaCache
+	}()
+
+	var key mataCacheKey
+	key[0] = 7
+	started := make(chan struct{})
+	release := make(chan struct{})
+	ownerDone := make(chan error, 1)
+
+	go func() {
+		_, err := dedupLoad(context.Background(), key, func() ([]byte, error) {
+			close(started)
+			<-release
+			return []byte("ok"), nil
+		})
+		ownerDone <- err
+	}()
+	<-started
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
+	defer cancel()
+	_, err := dedupLoad(ctx, key, func() ([]byte, error) {
+		return nil, errors.New("unexpected waiter load")
+	})
+	assert.ErrorIs(t, err, context.DeadlineExceeded)
+	assert.NotContains(t, err.Error(), "dedup load did not complete")
+
+	close(release)
+	assert.NoError(t, <-ownerDone)
+}
+
 func TestDedupLoadCleansUpAfterLoadCancel(t *testing.T) {
 	oldMetaCache := metaCache
 	metaCache = newMetaCache(fscache.ConstCapacity(1024))