diff --git a/core/kv/kv.go b/core/kv/kv.go index e2a1ad1..f287c92 100644 --- a/core/kv/kv.go +++ b/core/kv/kv.go @@ -50,3 +50,32 @@ type Batch interface { Close() error Count() int32 } + +// Snapshotter is an OPTIONAL capability. A backend that can serve a consistent, +// point-in-time read view implements it in addition to Store. Callers obtain the +// view by type-asserting a Store: sn, ok := store.(kv.Snapshotter). +type Snapshotter interface { + // Snapshot returns a read-only view frozen at the current committed state. + // Reads issued through it are unaffected by writes committed after the call. + // The caller MUST Close the returned Snapshot to release the resources it + // pins (see the Snapshot.Close contract). On error it returns (nil, err) — a + // literal untyped nil Snapshot, never a typed-nil pointer. It is safe to call + // concurrently with Store writes and other reads. + Snapshot() (Snapshot, error) +} + +// Snapshot is a read-only, point-in-time view over a Store. Its read methods +// mirror Store's (same key/value copy and callback slice-validity semantics), and +// a single Snapshot is safe for concurrent Get/Scan/ScanRange from multiple +// goroutines (each call opens its own reader/iterator). +type Snapshot interface { + Get(key []byte) ([]byte, error) + Scan(prefix []byte, cb func(key, value []byte) bool) error + ScanRange(begin, end []byte, cb func(key, value []byte) bool) error + // Close releases the view. It MUST be called (typically deferred) and MUST + // happen before the parent Store is closed. Close is idempotent and safe to + // call more than once (a second call is a no-op returning nil), consistent + // with pebblekv's existing repeat-safe Close (PebbleBatch.Close). Reads + // through the Snapshot after its Close are undefined; callers must not do that. + Close() error +} diff --git a/core/kv/pebblekv/db.go b/core/kv/pebblekv/db.go index da29425..b21b121 100644 --- a/core/kv/pebblekv/db.go +++ b/core/kv/pebblekv/db.go @@ -24,10 +24,27 @@ type pebbleStore interface { Delete(key []byte, opts *pebble.WriteOptions) error NewBatch() *pebble.Batch NewIter(o *pebble.IterOptions) (*pebble.Iterator, error) + NewSnapshot() *pebble.Snapshot Compact(start, end []byte, parallelize bool) error Close() error } +// var _ kv.Snapshotter = (*PebbleDB)(nil) asserts at compile time that PebbleDB +// implements the optional kv.Snapshotter capability (in addition to kv.Store). +var _ kv.Snapshotter = (*PebbleDB)(nil) + +// pebbleIterable and pebbleGetter are the minimal read surfaces the shared scan +// and get helpers need. Both the whole DB (*pebble.DB via pebbleStore) and a +// point-in-time snapshot (pebbleSnapshotReader) satisfy them, so PebbleDB and +// pebbleSnapshot share a single source of truth for the scan/get logic (notably +// the subtle keyUpperBound / prefix-HasPrefix bound, which has regressed before). +type pebbleIterable interface { + NewIter(o *pebble.IterOptions) (*pebble.Iterator, error) +} +type pebbleGetter interface { + Get(key []byte) ([]byte, io.Closer, error) +} + // PebbleDB is a Pebble-backed implementation of kv.Store. It wraps an // underlying Pebble database and tracks whether it has been closed. type PebbleDB struct { @@ -178,7 +195,12 @@ func (d *PebbleDB) GetIncrementalId(key []byte) (int, error) { return nextId, nil } -// Close closes the database +// Close closes the database. +// +// All snapshots obtained via Snapshot() MUST be Closed before Close is called: +// pebble's DB.Close returns a "leaked snapshots" error if any snapshot is still +// open. PebbleDB does not track or auto-close snapshots on Close — the ordering +// is the caller's contract (see kv.Snapshot.Close). func (d *PebbleDB) Close() error { if d.IsClosed() { return fmt.Errorf("database is closed") @@ -219,7 +241,15 @@ func (d *PebbleDB) Get(key []byte) ([]byte, error) { } // Read directly from the DB - value, closer, err := d.db.Get(key) + return getCopy(d.db, key) +} + +// getCopy reads key from r and returns a freshly-copied value (the source slice +// may be invalidated once the pebble Closer is closed), mapping pebble.ErrNotFound +// to (nil, nil). It is the single source of truth for the get semantics shared by +// PebbleDB.Get and pebbleSnapshot.Get. +func getCopy(r pebbleGetter, key []byte) ([]byte, error) { + value, closer, err := r.Get(key) if err == pebble.ErrNotFound { return nil, nil } @@ -283,11 +313,19 @@ func (d *PebbleDB) Scan(prefix []byte, cb func(key, value []byte) bool) error { return fmt.Errorf("database is closed") } + return scanPrefix(d.db, prefix, cb) +} + +// scanPrefix iterates every key of r that begins with prefix, invoking cb with +// each key/value (valid only for that call). Returning false from cb stops the +// scan. It is the single source of truth for the prefix-scan semantics shared by +// PebbleDB.Scan and pebbleSnapshot.Scan. +func scanPrefix(r pebbleIterable, prefix []byte, cb func(key, value []byte) bool) error { // Create an iterator with the prefix. The upper bound must be the prefix // successor (keyUpperBound), NOT append(prefix, 0xff): the latter excludes // any key of the form prefix+0xff+... — e.g. an inverted-index keyword // containing 0xff right after a search prefix — silently dropping it. - iter, err := d.db.NewIter(&pebble.IterOptions{ + iter, err := r.NewIter(&pebble.IterOptions{ LowerBound: prefix, UpperBound: keyUpperBound(prefix), }) @@ -324,8 +362,16 @@ func (d *PebbleDB) ScanRange(begin []byte, end []byte, cb func(key, value []byte return fmt.Errorf("database is closed") } + return scanRange(d.db, begin, end, cb) +} + +// scanRange iterates every key of r in [begin, end) (end exclusive), invoking cb +// with each key/value (valid only for that call). Returning false from cb stops +// the scan. It is the single source of truth for the range-scan semantics shared +// by PebbleDB.ScanRange and pebbleSnapshot.ScanRange. +func scanRange(r pebbleIterable, begin []byte, end []byte, cb func(key, value []byte) bool) error { // Create an iterator with the prefix - iter, err := d.db.NewIter(&pebble.IterOptions{ + iter, err := r.NewIter(&pebble.IterOptions{ LowerBound: begin, UpperBound: end, }) @@ -347,3 +393,64 @@ func (d *PebbleDB) ScanRange(begin []byte, end []byte, cb func(key, value []byte } return nil } + +// pebbleSnapshotReader is the read surface of a *pebble.Snapshot that +// pebbleSnapshot depends on. Declaring it as an interface (rather than using +// *pebble.Snapshot directly) lets tests substitute a fake to drive the snapshot +// read/close error branches, exactly as errPebbleStore does for PebbleDB. +type pebbleSnapshotReader interface { + Get(key []byte) ([]byte, io.Closer, error) + NewIter(o *pebble.IterOptions) (*pebble.Iterator, error) + Close() error +} + +// pebbleSnapshot is a read-only, point-in-time kv.Snapshot backed by a +// *pebble.Snapshot. Its Get/Scan/ScanRange delegate to the same shared helpers +// as PebbleDB, so their semantics are byte-identical. A pebbleSnapshot has no +// independent closed-state: closing the parent DB while it is open is a caller +// contract violation (see PebbleDB.Close / kv.Snapshot.Close). +type pebbleSnapshot struct { + snap pebbleSnapshotReader +} + +func (s *pebbleSnapshot) Get(key []byte) ([]byte, error) { + return getCopy(s.snap, key) +} + +func (s *pebbleSnapshot) Scan(prefix []byte, cb func(key, value []byte) bool) error { + return scanPrefix(s.snap, prefix, cb) +} + +func (s *pebbleSnapshot) ScanRange(begin, end []byte, cb func(key, value []byte) bool) error { + return scanRange(s.snap, begin, end, cb) +} + +// Close releases the snapshot. It is idempotent: the reader field is nil'd +// UNCONDITIONALLY on the first call (before/regardless of the underlying Close's +// result), so a later Close is a safe no-op returning nil and never re-invokes +// the underlying Close — which would panic on a raw pebble.Snapshot double-close. +// Consistent with pebblekv's repeat-safe PebbleBatch.Close. +func (s *pebbleSnapshot) Close() error { + snap := s.snap + s.snap = nil + if snap == nil { + return nil + } + return snap.Close() +} + +// Snapshot returns a consistent, point-in-time read view of the DB, satisfying +// the optional kv.Snapshotter capability. On a closed DB it returns a non-nil +// error together with a literal-nil kv.Snapshot (never a typed-nil +// *pebbleSnapshot, which would defeat a caller's sn == nil check). The IsClosed +// guard is required, not cosmetic: pebble's DB.NewSnapshot panics with ErrClosed +// on a closed *pebble.DB (and Close nils d.db), so the guard must return the +// error before ever touching d.db — exactly like Get/Scan/ScanRange. It is safe +// to call concurrently with writes and other reads. +func (d *PebbleDB) Snapshot() (kv.Snapshot, error) { + if d.IsClosed() { + return nil, fmt.Errorf("database is closed") + } + + return &pebbleSnapshot{snap: d.db.NewSnapshot()}, nil +} diff --git a/core/kv/pebblekv/pebble_test.go b/core/kv/pebblekv/pebble_test.go index a9cdc9b..7193dbd 100644 --- a/core/kv/pebblekv/pebble_test.go +++ b/core/kv/pebblekv/pebble_test.go @@ -4,12 +4,16 @@ import ( "errors" "fmt" "io" + "reflect" + "sync" "sync/atomic" "testing" "time" pebbledb "github.com/cockroachdb/pebble" + "github.com/codetrek/haystack/core/kv" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) // errBatchWriter is a mock pebbleBatchWriter that returns errors from every operation. @@ -562,6 +566,10 @@ func (e *errPebbleStore) NewBatch() *pebbledb.Batch { return nil } func (e *errPebbleStore) NewIter(o *pebbledb.IterOptions) (*pebbledb.Iterator, error) { return nil, e.err } + +// NewSnapshot is never dereferenced on errPebbleStore's paths (no test drives +// PebbleDB.Snapshot through this fake); it exists only to satisfy pebbleStore. +func (e *errPebbleStore) NewSnapshot() *pebbledb.Snapshot { return nil } func (e *errPebbleStore) Compact(start, end []byte, parallelize bool) error { return e.err } func (e *errPebbleStore) Close() error { return nil } @@ -606,3 +614,441 @@ func TestDB_ScanRange_StopEarly(t *testing.T) { }) assert.Equal(t, 2, count) } + +// --------------------------------------------------------------------------- +// kv.Snapshotter / kv.Snapshot (pebbleSnapshot) tests +// --------------------------------------------------------------------------- + +// openSnapTestDB opens a fresh pebblekv store in a temp dir for the snapshot +// tests. It fails the test (not merely records) on error so callers never +// dereference a nil Store. +func openSnapTestDB(t *testing.T) kv.Store { + t.Helper() + db, err := Open(t.TempDir()+"/snapdb", 4*1024*1024) + require.NoError(t, err) + require.NotNil(t, db) + return db +} + +// mustSnapshot obtains a snapshot via the OPTIONAL kv.Snapshotter capability +// (type-asserted from kv.Store, exactly how a caller acquires one) and fails the +// test if the store does not implement it or Snapshot errors. +func mustSnapshot(t *testing.T, db kv.Store) kv.Snapshot { + t.Helper() + ss, ok := db.(kv.Snapshotter) + require.True(t, ok, "pebblekv store must implement kv.Snapshotter") + snap, err := ss.Snapshot() + require.NoError(t, err) + require.NotNil(t, snap) + return snap +} + +// errSnapReader implements pebbleSnapshotReader and returns the injected error +// from every method (Get/NewIter/Close). It lets the snapshot's read/close error +// branches be driven directly via pebbleSnapshot{snap: errSnapReader{err}} — the +// errPebbleStore fake cannot, because its Close returns nil. Value receivers so +// the field can be nil'd by pebbleSnapshot.Close without aliasing surprises. +type errSnapReader struct { + err error +} + +func (e errSnapReader) Get(key []byte) ([]byte, io.Closer, error) { + return nil, nil, e.err +} +func (e errSnapReader) NewIter(o *pebbledb.IterOptions) (*pebbledb.Iterator, error) { + return nil, e.err +} +func (e errSnapReader) Close() error { + return e.err +} + +// TestSnapshot_Consistency pins contract §4.1: reads through a snapshot observe +// exactly the state committed when Snapshot() returned. Writes committed after +// (overwrite, new key, delete) are invisible to it, while the live DB reflects +// them. A snapshot that read the live DB would fail every assertion here. +func TestSnapshot_Consistency(t *testing.T) { + db := openSnapTestDB(t) + defer db.Close() + + // Seed pre-snapshot state: k=v1 and an "old" key we will later delete. + require.NoError(t, db.Put([]byte("k"), []byte("v1"))) + require.NoError(t, db.Put([]byte("old"), []byte("kept"))) + + snap := mustSnapshot(t, db) + defer snap.Close() + + // Mutate the DB AFTER the snapshot: overwrite k, add a new key, delete old. + require.NoError(t, db.Put([]byte("k"), []byte("v2"))) + require.NoError(t, db.Put([]byte("new"), []byte("fresh"))) + require.NoError(t, db.Delete([]byte("old"))) + + // The snapshot still sees the frozen state. + got, err := snap.Get([]byte("k")) + assert.NoError(t, err) + assert.Equal(t, []byte("v1"), got, "snapshot must still read the pre-write value") + + got, err = snap.Get([]byte("new")) + assert.NoError(t, err) + assert.Nil(t, got, "snapshot must NOT see a key added after it was taken") + + got, err = snap.Get([]byte("old")) + assert.NoError(t, err) + assert.Equal(t, []byte("kept"), got, "snapshot must still see a key deleted after it was taken") + + // The live DB reflects the post-snapshot writes. + got, err = db.Get([]byte("k")) + assert.NoError(t, err) + assert.Equal(t, []byte("v2"), got) + got, err = db.Get([]byte("new")) + assert.NoError(t, err) + assert.Equal(t, []byte("fresh"), got) + got, err = db.Get([]byte("old")) + assert.NoError(t, err) + assert.Nil(t, got) +} + +// TestSnapshot_ScanIsolation pins contract §4.1 for Scan/ScanRange: after +// mutations under the prefix (overwrite, add, delete), the snapshot yields the +// EXACT original key/value set — not a count, so a wrong impl that saw the +// mutated value or the added/removed key is caught. +func TestSnapshot_ScanIsolation(t *testing.T) { + db := openSnapTestDB(t) + defer db.Close() + + original := map[string]string{ + "p:a": "1", + "p:b": "2", + "p:c": "3", + } + for k, v := range original { + require.NoError(t, db.Put([]byte(k), []byte(v))) + } + + snap := mustSnapshot(t, db) + defer snap.Close() + + // Mutate under the prefix after the snapshot. + require.NoError(t, db.Put([]byte("p:a"), []byte("changed"))) // overwrite + require.NoError(t, db.Put([]byte("p:d"), []byte("4"))) // add + require.NoError(t, db.Delete([]byte("p:b"))) // delete + + gotScan := map[string]string{} + err := snap.Scan([]byte("p:"), func(k, v []byte) bool { + gotScan[string(k)] = string(v) + return true + }) + assert.NoError(t, err) + assert.Equal(t, original, gotScan, "Scan through the snapshot must yield the frozen set") + + // ScanRange over [p:, p;) spans exactly the p: keys; must match likewise. + gotRange := map[string]string{} + err = snap.ScanRange([]byte("p:"), []byte("p;"), func(k, v []byte) bool { + gotRange[string(k)] = string(v) + return true + }) + assert.NoError(t, err) + assert.Equal(t, original, gotRange, "ScanRange through the snapshot must yield the frozen set") +} + +// TestSnapshot_Scan_PrefixFollowedByFF mirrors TestDB_Scan_PrefixFollowedByFF +// through the snapshot path (contract §4.4 / R3): every key with the prefix must +// be scanned, including keys whose first byte after the prefix is 0xff. Asserts +// the EXACT key set (stronger than a count) so a regressed keyUpperBound bound in +// the shared scanPrefix helper is caught on the snapshot path too. +func TestSnapshot_Scan_PrefixFollowedByFF(t *testing.T) { + db := openSnapTestDB(t) + defer db.Close() + + prefix := []byte("p|") + keys := [][]byte{ + []byte("p|a"), + {'p', '|', 0xff}, + {'p', '|', 0xff, 'x'}, + {'p', '|', 0xff, 0xff, 'z'}, + } + for i, k := range keys { + require.NoError(t, db.Put(k, []byte{byte(i)})) + } + require.NoError(t, db.Put([]byte("q|other"), []byte("x"))) // outside the prefix + + snap := mustSnapshot(t, db) + defer snap.Close() + + want := map[string]bool{} + for _, k := range keys { + want[string(k)] = true + } + + got := map[string]bool{} + err := snap.Scan(prefix, func(k, _ []byte) bool { + got[string(append([]byte{}, k...))] = true + return true + }) + assert.NoError(t, err) + assert.Equal(t, want, got, + "snapshot Scan must yield exactly the prefix keys, including those with 0xff right after the prefix") +} + +// TestSnapshot_Get_CopyAndNotFound pins contract §4.3: snapshot Get returns a +// fresh copy (mutating it must not corrupt a subsequent read) and maps a missing +// key to (nil, nil). +func TestSnapshot_Get_CopyAndNotFound(t *testing.T) { + db := openSnapTestDB(t) + defer db.Close() + + require.NoError(t, db.Put([]byte("k"), []byte("orig"))) + + snap := mustSnapshot(t, db) + defer snap.Close() + + got, err := snap.Get([]byte("k")) + assert.NoError(t, err) + assert.Equal(t, []byte("orig"), got) + + // Mutating the returned slice must not affect a re-Get. + for i := range got { + got[i] = 'X' + } + again, err := snap.Get([]byte("k")) + assert.NoError(t, err) + assert.Equal(t, []byte("orig"), again, "Get must return a copy; mutating it must not corrupt the view") + + // Absent key → (nil, nil). + absent, err := snap.Get([]byte("missing")) + assert.NoError(t, err) + assert.Nil(t, absent) +} + +// TestSnapshot_Scan_StopEarly pins contract §4.4: returning false from the cb +// stops iteration for both Scan and ScanRange through the snapshot. +func TestSnapshot_Scan_StopEarly(t *testing.T) { + db := openSnapTestDB(t) + defer db.Close() + + for _, k := range []string{"k:1", "k:2", "k:3"} { + require.NoError(t, db.Put([]byte(k), []byte("v"))) + } + + snap := mustSnapshot(t, db) + defer snap.Close() + + count := 0 + err := snap.Scan([]byte("k:"), func(k, v []byte) bool { + count++ + return count < 2 // stop after 2 + }) + assert.NoError(t, err) + assert.Equal(t, 2, count) + + rcount := 0 + err = snap.ScanRange([]byte("k:1"), []byte("k:9"), func(k, v []byte) bool { + rcount++ + return rcount < 2 // stop after 2 + }) + assert.NoError(t, err) + assert.Equal(t, 2, rcount) +} + +// TestSnapshot_ConcurrentReads pins contract §4.2(a)+(b): a single snapshot is +// read by N goroutines (Get + Scan) WHILE other goroutines write the DB +// (Put/Delete/add). Every read must observe the frozen state, and the run must +// be race-clean under -race. A wrong impl (shared iterator, or reading the live +// DB) either races or observes a mutated value. +func TestSnapshot_ConcurrentReads(t *testing.T) { + db := openSnapTestDB(t) + defer db.Close() + + const nKeys = 50 + frozen := map[string]string{} + for i := 0; i < nKeys; i++ { + k := fmt.Sprintf("p:%03d", i) + v := fmt.Sprintf("v%03d", i) + frozen[k] = v + require.NoError(t, db.Put([]byte(k), []byte(v))) + } + require.NoError(t, db.Put([]byte("solo"), []byte("frozen-solo"))) + + snap := mustSnapshot(t, db) + defer snap.Close() + + // Writers churn the DB under the same prefix + a growing "extra:" set. + stop := make(chan struct{}) + var writers sync.WaitGroup + for w := 0; w < 4; w++ { + writers.Add(1) + go func() { + defer writers.Done() + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + _ = db.Put([]byte(fmt.Sprintf("p:%03d", i%nKeys)), []byte("MUTATED")) + _ = db.Put([]byte("solo"), []byte("MUTATED")) + _ = db.Delete([]byte(fmt.Sprintf("p:%03d", (i+1)%nKeys))) + _ = db.Put([]byte(fmt.Sprintf("extra:%d", i)), []byte("x")) + } + }() + } + + // Readers repeatedly read the snapshot; each read must match the frozen set. + var readers sync.WaitGroup + errCh := make(chan error, 32) + for r := 0; r < 4; r++ { + readers.Add(1) + go func() { + defer readers.Done() + for iter := 0; iter < 200; iter++ { + got, err := snap.Get([]byte("solo")) + if err != nil { + errCh <- fmt.Errorf("snapshot Get: %w", err) + return + } + if string(got) != "frozen-solo" { + errCh <- fmt.Errorf("snapshot Get(solo)=%q, want frozen-solo", got) + return + } + seen := map[string]string{} + if err := snap.Scan([]byte("p:"), func(k, v []byte) bool { + seen[string(k)] = string(v) + return true + }); err != nil { + errCh <- fmt.Errorf("snapshot Scan: %w", err) + return + } + if !reflect.DeepEqual(seen, frozen) { + errCh <- fmt.Errorf("snapshot Scan saw a mutated set (%d keys)", len(seen)) + return + } + } + }() + } + + readers.Wait() + close(stop) + writers.Wait() + close(errCh) + for err := range errCh { + t.Error(err) + } +} + +// TestSnapshot_CloseBeforeDBClose pins contract §4.7 (positive): with the +// snapshot Closed first, the parent DB.Close returns nil. +func TestSnapshot_CloseBeforeDBClose(t *testing.T) { + db := openSnapTestDB(t) + require.NoError(t, db.Put([]byte("k"), []byte("v"))) + + snap := mustSnapshot(t, db) + assert.NoError(t, snap.Close()) + + assert.NoError(t, db.Close(), "DB.Close must succeed once the snapshot is closed") +} + +// TestSnapshot_OpenSnapshotBlocksDBClose pins contract §4.7 / R2 (negative): +// closing the DB while a snapshot is still open returns a non-nil error. It +// wraps pebble's "leaked snapshots" error; asserting on that substring is an +// INTENTIONAL, documented coupling to the pebble behavior the ordering contract +// rests on, so a pebble upgrade that silently weakened it is caught. +func TestSnapshot_OpenSnapshotBlocksDBClose(t *testing.T) { + db := openSnapTestDB(t) + require.NoError(t, db.Put([]byte("k"), []byte("v"))) + + snap := mustSnapshot(t, db) + + err := db.Close() + assert.Error(t, err, "DB.Close with an open snapshot must return an error") + assert.Contains(t, err.Error(), "leaked", + "DB.Close with an open snapshot must surface pebble's leaked-snapshots error (documented coupling)") + + // Release the snapshot to clean up. + assert.NoError(t, snap.Close()) +} + +// TestSnapshot_CloseIdempotent pins contract §4.6: a second Close is a safe +// no-op returning nil and never panics (defends the defer+explicit-Close +// pattern from pebble's double-close panic). +func TestSnapshot_CloseIdempotent(t *testing.T) { + db := openSnapTestDB(t) + defer db.Close() + + snap := mustSnapshot(t, db) + + assert.NoError(t, snap.Close()) + assert.NotPanics(t, func() { + assert.NoError(t, snap.Close(), "second Close must be a no-op returning nil") + }) +} + +// TestSnapshot_OnClosedStore pins contract §4.8: Snapshot() on a closed store +// returns a non-nil error AND a LITERAL-nil kv.Snapshot (not a typed-nil +// *pebbleSnapshot, which would defeat a caller's sn == nil check). +func TestSnapshot_OnClosedStore(t *testing.T) { + db := openSnapTestDB(t) + require.NoError(t, db.Close()) + + ss, ok := db.(kv.Snapshotter) + require.True(t, ok) + + snap, err := ss.Snapshot() + assert.Error(t, err) + assert.True(t, snap == nil, "Snapshot on a closed store must return a literal-nil kv.Snapshot, not a typed nil") +} + +// --------------------------------------------------------------------------- +// T4: underlying-error propagation via the errSnapReader fake. These drive the +// error branches of getCopy/scanPrefix/scanRange and pebbleSnapshot.Close that a +// real *pebble.Snapshot cannot be made to hit, closing the deferred go-cov +// CRITICAL markers. +// --------------------------------------------------------------------------- + +// TestSnapshot_GetError: a snapshot Get propagates the reader's Get error +// (getCopy's non-NotFound branch: "failed to get data"). +func TestSnapshot_GetError(t *testing.T) { + injectedErr := errors.New("snapshot get failed") + s := &pebbleSnapshot{snap: errSnapReader{err: injectedErr}} + + _, err := s.Get([]byte("k")) + assert.Error(t, err) + assert.Contains(t, err.Error(), "failed to get data") + assert.Contains(t, err.Error(), "snapshot get failed") +} + +// TestSnapshot_ScanError: a snapshot Scan propagates the reader's NewIter error +// (scanPrefix's "failed to create iterator" branch). +func TestSnapshot_ScanError(t *testing.T) { + injectedErr := errors.New("snapshot newiter failed") + s := &pebbleSnapshot{snap: errSnapReader{err: injectedErr}} + + err := s.Scan([]byte("p:"), func(k, v []byte) bool { return true }) + assert.Error(t, err) + assert.Contains(t, err.Error(), "failed to create iterator") + assert.Contains(t, err.Error(), "snapshot newiter failed") +} + +// TestSnapshot_ScanRangeError: a snapshot ScanRange propagates the reader's +// NewIter error (scanRange's "failed to create iterator" branch). +func TestSnapshot_ScanRangeError(t *testing.T) { + injectedErr := errors.New("snapshot newiter failed") + s := &pebbleSnapshot{snap: errSnapReader{err: injectedErr}} + + err := s.ScanRange([]byte("a"), []byte("z"), func(k, v []byte) bool { return true }) + assert.Error(t, err) + assert.Contains(t, err.Error(), "failed to create iterator") + assert.Contains(t, err.Error(), "snapshot newiter failed") +} + +// TestSnapshot_CloseError: the FIRST Close propagates the reader's Close error +// (pebbleSnapshot.Close returns the captured reader's Close result verbatim). +func TestSnapshot_CloseError(t *testing.T) { + injectedErr := errors.New("snapshot close failed") + s := &pebbleSnapshot{snap: errSnapReader{err: injectedErr}} + + err := s.Close() + assert.ErrorIs(t, err, injectedErr) + + // Idempotency still holds after an erroring Close: the reader was nil'd + // unconditionally, so a second Close is a no-op returning nil. + assert.NoError(t, s.Close()) +}