store: fix error handling in decodePostings (#6650)

Signed-off-by: Michael Hoffmann <mhoffm@posteo.de>

CHANGELOG.md

@@ -12,6 +12,8 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 
 ### Fixed
 
+- [#6650](https://github.com/thanos-io/thanos/pull/6650) Store: fix error handling in decodePostings
+
 ### Added
 
 - [#6605](https://github.com/thanos-io/thanos/pull/6605) Query Frontend: Support vertical sharding binary expression with metric name when no matching labels specified.

pkg/store/bucket.go

@@ -2864,14 +2864,13 @@ func (r *bucketIndexReader) decodeCachedPostings(b []byte) (index.Postings, []fu
 	)
 	if isDiffVarintSnappyEncodedPostings(b) || isDiffVarintSnappyStreamedEncodedPostings(b) {
 		s := time.Now()
-		clPostings, err := decodePostings(b)
+		l, err = decodePostings(b)
 		r.stats.cachedPostingsDecompressions += 1
 		r.stats.CachedPostingsDecompressionTimeSum += time.Since(s)
 		if err != nil {
 			r.stats.cachedPostingsDecompressionErrors += 1
 		} else {
-			closeFns = append(closeFns, clPostings.close)
-			l = clPostings
+			closeFns = append(closeFns, l.(closeablePostings).close)
 		}
 	} else {
 		_, l, err = r.dec.Postings(b)

pkg/store/bucket_test.go

@@ -3330,3 +3330,15 @@ func TestExpandedPostingsRace(t *testing.T) {
 		wg.Wait()
 	}
 }
+
+func TestBucketIndexReader_decodeCachedPostingsErrors(t *testing.T) {
+	bir := bucketIndexReader{stats: &queryStats{}}
+	t.Run("should return error on broken cached postings without snappy prefix", func(t *testing.T) {
+		_, _, err := bir.decodeCachedPostings([]byte("foo"))
+		testutil.NotOk(t, err)
+	})
+	t.Run("should return error on broken cached postings with snappy prefix", func(t *testing.T) {
+		_, _, err := bir.decodeCachedPostings(append([]byte(codecHeaderSnappy), []byte("foo")...))
+		testutil.NotOk(t, err)
+	})
+}