From f9cc10c97a505c1a5aa7fd4369d8a2c31714292b Mon Sep 17 00:00:00 2001
From: Radu Berinde <radu@cockroachlabs.com>
Date: Tue, 20 Feb 2024 17:35:50 -0800
Subject: [PATCH] metamorphic: special handling of assertion errors

Check for assertion errors when recording history and display their
stacktrace and panic. This helps with debugging.

We also fix some cases where we return an assertion error when we are
checking an API input; assertion errors should indicate a bug inside
Pebble.
---
 ingest.go                            |  8 ++++----
 metamorphic/history.go               |  7 +++++++
 sstable/prefix_replacing_iterator.go | 13 +++++++++----
 3 files changed, 20 insertions(+), 8 deletions(-)

diff --git a/ingest.go b/ingest.go
index 722b037f627..b37cc4b3d56 100644
--- a/ingest.go
+++ b/ingest.go
@@ -491,14 +491,14 @@ func ingestSortAndVerify(cmp Compare, lr ingestLoadResult, exciseSpan KeyRange)
 	// fit within the exciseSpan.
 	for _, f := range lr.shared {
 		if !exciseSpan.Contains(cmp, f.Smallest) || !exciseSpan.Contains(cmp, f.Largest) {
-			return errors.AssertionFailedf("pebble: shared file outside of excise span, span [%s-%s), file = %s", exciseSpan.Start, exciseSpan.End, f.String())
+			return errors.Newf("pebble: shared file outside of excise span, span [%s-%s), file = %s", exciseSpan.Start, exciseSpan.End, f.String())
 		}
 	}
 	if len(lr.external) > 0 {
 		if len(lr.local) > 0 || len(lr.shared) > 0 {
 			// Currently we only support external ingests on their own. If external
 			// files are present alongside local/shared files, return an error.
-			return errors.AssertionFailedf("pebble: external files cannot be ingested atomically alongside other types of files")
+			return errors.Newf("pebble: external files cannot be ingested atomically alongside other types of files")
 		}
 		// Sort according to the smallest key.
 		slices.SortFunc(lr.external, func(a, b ingestExternalMeta) int {
@@ -506,7 +506,7 @@ func ingestSortAndVerify(cmp Compare, lr ingestLoadResult, exciseSpan KeyRange)
 		})
 		for i := 1; i < len(lr.external); i++ {
 			if sstableKeyCompare(cmp, lr.external[i-1].Largest, lr.external[i].Smallest) >= 0 {
-				return errors.AssertionFailedf("pebble: external sstables have overlapping ranges")
+				return errors.Newf("pebble: external sstables have overlapping ranges")
 			}
 		}
 		return nil
@@ -522,7 +522,7 @@ func ingestSortAndVerify(cmp Compare, lr ingestLoadResult, exciseSpan KeyRange)
 
 	for i := 1; i < len(lr.local); i++ {
 		if sstableKeyCompare(cmp, lr.local[i-1].Largest, lr.local[i].Smallest) >= 0 {
-			return errors.AssertionFailedf("pebble: local ingestion sstables have overlapping ranges")
+			return errors.Newf("pebble: local ingestion sstables have overlapping ranges")
 		}
 	}
 	if len(lr.shared) == 0 {
diff --git a/metamorphic/history.go b/metamorphic/history.go
index cac1147b390..3829feef58c 100644
--- a/metamorphic/history.go
+++ b/metamorphic/history.go
@@ -148,6 +148,13 @@ type historyRecorder struct {
 
 // Recordf records the results of a single operation.
 func (h historyRecorder) Recordf(format string, args ...interface{}) {
+	// Check for assertion errors.
+	for _, a := range args {
+		if err, ok := a.(error); ok && errors.IsAssertionFailure(err) {
+			fmt.Fprintf(os.Stderr, "%+v", err)
+			panic(err)
+		}
+	}
 	h.history.Recordf(h.op, format, args...)
 	// If the history recorder was configured with an additional record func,
 	// invoke it. This can be used to collect the per-operation output when
diff --git a/sstable/prefix_replacing_iterator.go b/sstable/prefix_replacing_iterator.go
index cf223c9879d..d3d49ab899b 100644
--- a/sstable/prefix_replacing_iterator.go
+++ b/sstable/prefix_replacing_iterator.go
@@ -41,8 +41,13 @@ type prefixReplacingIterator struct {
 	empty bool
 }
 
-var errInputPrefixMismatch = errors.New("key argument does not have prefix required for replacement")
-var errOutputPrefixMismatch = errors.New("key returned does not have prefix required for replacement")
+func errInputPrefixMismatch() error {
+	return errors.AssertionFailedf("key argument does not have prefix required for replacement")
+}
+
+func errOutputPrefixMismatch() error {
+	return errors.AssertionFailedf("key returned does not have prefix required for replacement")
+}
 
 var _ Iterator = (*prefixReplacingIterator)(nil)
 
@@ -101,7 +106,7 @@ func (p *prefixReplacingIterator) rewriteResult(
 		return k, v
 	}
 	if !bytes.HasPrefix(k.UserKey, p.contentPrefix) {
-		p.err = errOutputPrefixMismatch
+		p.err = errOutputPrefixMismatch()
 		if invariants.Enabled {
 			panic(p.err)
 		}
@@ -280,7 +285,7 @@ func newPrefixReplacingFragmentIterator(
 
 func (p *prefixReplacingFragmentIterator) rewriteArg(key []byte) ([]byte, error) {
 	if !bytes.HasPrefix(key, p.syntheticPrefix) {
-		return nil, errInputPrefixMismatch
+		return nil, errInputPrefixMismatch()
 	}
 	p.arg = append(p.arg[:len(p.contentPrefix)], key[len(p.syntheticPrefix):]...)
 	return p.arg, nil