Snapshot/Ingest race can drop a key visible at NewSnapshot time

## Summary

A regular `DB.NewSnapshot()` racing concurrently with an `Ingest` (without excise — the excise path has separate machinery) can observe data loss. A key that existed before the ingest can disappear from the snapshot's view (`Get` returns `ErrNotFound`), even though it was visible at the moment the snapshot was created.

## Race

The commit pipeline's `AllocateSeqNum` for ingest does (`commit.go`):

```
1. logSeqNum.Add(count) // ingest gets seqnums [N, N+count-1]
2. wait until visibleSeqNum == N
3. prepare(...) // under commitPipeline.mu
4. apply(...) // -> d.ingestApply: takes d.mu, runs
// UpdateVersionLocked, installs the
// new sstables into the version,
// calls maybeScheduleCompaction,
// releases d.mu
5. publish(...) // bumps visibleSeqNum to N+count
```

Between step 4 returning and step 5 completing, `d.mu` is **not held** and `visibleSeqNum` is still `N`, but the current `Version` already contains the ingested sstables with `LargestSeqNum >= N`. This breaks the invariant that compactions rely on for snapshot safety.

The standard safety argument (`NewSnapshot` reads `visibleSeqNum` under `d.mu`; compactions read `d.mu.snapshots` under `d.mu`; both atomic, so for any in-flight compaction `V_c >= L` where `L` is the max input seqnum) holds for regular writes — keys go through the memtable and `visibleSeqNum` is bumped before the memtable is flushed into an sstable. For ingest the inequality flips: input sstables can have seqnums up to `L = N+count-1` while `V_c == N`.

### Bad interleaving

- T1: ingest's `ingestApply` returns; `d.mu` released; version contains the new sst at seqnum `N`; `visibleSeqNum == N`.
- T2: a compaction goroutine (e.g., the one scheduled by `maybeScheduleCompaction` inside `ingestApply`) acquires `d.mu`, runs into `runDefaultTableCompaction`, captures `snapshots := d.mu.snapshots.toSlice()`, releases `d.mu`, starts I/O. The compaction's inputs include the freshly-ingested L5 sst and the overlapping L6 sst with the prior version of the key.
- T3: a goroutine calls `d.NewSnapshot()` — acquires `d.mu`, reads `visibleSeqNum == N`, registers `S = N`.
- T4: the compaction iterator processes the user-key. The stripe `(d, N]` (between old `d < N` and new at `N`) contains no snapshot in the compaction's captured list, so the compaction elides `d`. The surviving output is `"k"@N`.
- T5: ingest's `publish` bumps `visibleSeqNum` to `N+count`.
- T6: `snap.Get("k")` at `S = N` filters out `"k"@N` (visibility is `seqNum < snapshot`, so `N < N` is false) and the old version is gone → `ErrNotFound`.

Note: the "bottommost snapshot stripe" optimization in `internal/compact/iterator.go` rewrites surviving last-stripe keys to `SeqNumZero` when writing to the bottom LSM level. With no concurrent snapshots, the surviving `"k"@N` is rewritten to `"k"@0`, which the racing snapshot at `S=N` still sees as the post-ingest value `"new"` — masking the bug as a benign "snapshot effectively raced past the ingest" outcome. Keeping a long-lived snapshot below the ingested seqnum disables the rewrite and exposes the real `ErrNotFound`.

## Why prior fixes don't cover this

- The `NewSnapshot` race fix in `ae9095403 db: fix race in NewSnapshot` (June 2020) addressed reads happening before the snapshot was registered. The fix moves the `visibleSeqNum.Load()` under `d.mu`. That's necessary, but its safety argument assumes `V_c >= L` for any in-flight compaction — which fails for ingest as described above.
- `TestNewSnapshotRace` covers the Set+Flush variant, not Ingest.
- The `EventuallyFileOnlySnapshot` / `ongoingExcises` machinery in `snapshot.go` (e.g., `makeEventuallyFileOnlySnapshot.tryGetSnapshotSeqNum`) addresses an EFOS-specific variant of this race only for ingest-and-**excise**, and only for EFOS — not for plain `NewSnapshot` racing a plain ingest.
- The metamorphic test does not exercise it: both `ingestOp.receiver()` and `newSnapshotOp.receiver()` return `dbID`, and `meta.go`'s parallel executor partitions ops by `hashThread(receiver, numThreads)`. So on a single DB, ingests and snapshots run serially on the same goroutine.

## Reproducer

Adds two private test hooks and a deterministic test. The test fails on `master` with `pebble: not found` for the racing snapshot's `Get`.

### Hook plumbing

```go
// options.go — added two private hooks.
testingAfterIngestApplyFunc func()
testingAfterCompactionSnapshotGrabFunc func()
```

```go
// ingest.go — inside the apply closure of the IngestAndExcise path,
// just after d.ingestApply returns successfully:
if err == nil && d.opts.private.testingAfterIngestApplyFunc != nil {
d.opts.private.testingAfterIngestApplyFunc()
}
```

```go
// compaction.go — inside runDefaultTableCompaction, after the snapshot
// list is captured and d.mu has been released:
snapshots := d.mu.snapshots.toSlice()
d.mu.Unlock()
defer d.mu.Lock()
if d.opts.private.testingAfterCompactionSnapshotGrabFunc != nil {
d.opts.private.testingAfterCompactionSnapshotGrabFunc()
}
result := d.compactAndWrite(jobID, c, snapshots)
```

### Test

```go
// TestNewSnapshotIngestRace exercises the race window in which an ingest has
// installed its sstables into the current version (via DB.ingestApply) but has
// not yet bumped visibleSeqNum (via commitPipeline.publish).
func TestNewSnapshotIngestRace(t *testing.T) {
defer leaktest.AfterTest(t)()

var (
afterIngestApply = make(chan struct{})
releaseIngest = make(chan struct{})
afterCompactionSnapshotGrab = make(chan struct{})
releaseCompaction = make(chan struct{})
hooksEnabled atomic.Bool
ingestHookFired atomic.Bool
compactionHookFired atomic.Bool
)

o := &Options{
FS: vfs.NewMem(),
DisableAutomaticCompactions: true,
FormatMajorVersion: FormatNewest,
}
o.private.testingAfterIngestApplyFunc = func() {
if !hooksEnabled.Load() {
return
}
if ingestHookFired.CompareAndSwap(false, true) {
afterIngestApply <- struct{}{}
<-releaseIngest
}
}
o.private.testingAfterCompactionSnapshotGrabFunc = func() {
if !hooksEnabled.Load() {
return
}
if compactionHookFired.CompareAndSwap(false, true) {
afterCompactionSnapshotGrab <- struct{}{}
<-releaseCompaction
}
}

d, err := Open("", o)
require.NoError(t, err)
defer func() { require.NoError(t, d.Close()) }()

// Long-lived "barrier" snapshot keeps the L5->L6 compaction out of the
// bottommost snapshot stripe, disabling the SeqNumZero rewrite on the
// surviving "k"@new. Without this, the racing snapshot would observe
// "new" instead of NotFound.
barrier := d.NewSnapshot()
defer func() { require.NoError(t, barrier.Close()) }()

require.NoError(t, d.Set([]byte("k"), []byte("old"), nil))
require.NoError(t, d.Flush())
require.NoError(t, d.Compact(context.Background(), []byte("a"), []byte("z"), false))

hooksEnabled.Store(true)

f, err := o.FS.Create("ext", vfs.WriteCategoryUnspecified)
require.NoError(t, err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f),
d.opts.MakeWriterOptions(0, d.TableFormat()))
require.NoError(t, w.Set([]byte("k"), []byte("new")))
require.NoError(t, w.Close())

// Ingest blocks after apply, before publish: version updated,
// visibleSeqNum still N.
ingestDone := make(chan struct{})
go func() {
defer close(ingestDone)
require.NoError(t, d.Ingest(context.Background(), []string{"ext"}))
}()
<-afterIngestApply

// L5->L6 compaction pulls in both ssts; blocks after grabbing snapshot list.
compactDone := make(chan struct{})
go func() {
defer close(compactDone)
require.NoError(t, d.Compact(context.Background(), []byte("a"), []byte("z"), false))
}()
<-afterCompactionSnapshotGrab

// visibleSeqNum is still N; snapshot is registered at S=N AFTER the
// compaction has already captured its (barrier-only) snapshot list.
snap := d.NewSnapshot()

close(releaseCompaction)
<-compactDone

close(releaseIngest)
<-ingestDone

v, closer, getErr := snap.Get([]byte("k"))
var observed string
if getErr == nil {
observed = string(v)
_ = closer.Close()
}
require.NoError(t, snap.Close())

require.NoError(t, getErr, "snapshot should still see the key (saw NotFound — data loss)")
require.Contains(t, []string{"old", "new"}, observed,
"snapshot should observe either pre- or post-ingest value")
}
```

### Observed outcome

```
=== RUN TestNewSnapshotIngestRace
snapshot_test.go:596:
Error Trace: .../snapshot_test.go:596
Error: Received unexpected error:
pebble: not found
Test: TestNewSnapshotIngestRace
Messages: snapshot should still see the key (saw NotFound — data loss)
--- FAIL: TestNewSnapshotIngestRace (0.01s)
```

## Suggested fix directions

The cleanest fix is to close the apply→publish window so the version install and the `visibleSeqNum` bump are atomic with respect to `d.mu`. Options:

1. Hold `d.mu` across both `apply` and `publish` for the ingest's batch. Today `commit.AllocateSeqNum` releases `commitPipeline.mu` before `apply` and never takes `d.mu` itself; the ingest's `d.ingestApply` takes `d.mu` internally and releases it before returning.
2. Bump `visibleSeqNum` from inside `ingestApply` (after `UpdateVersionLocked`) while still under `d.mu`, and skip the corresponding work in `publish`. Requires care so that other pending batches in the commit queue still get published in order.
3. Have `runDefaultTableCompaction` exclude inputs whose `LargestSeqNum >= visibleSeqNum` from its captured set, or fail/retry the compaction if it observes such inputs. Tighter but more invasive in the picker.

Happy to take a stab once a direction is settled.

Jira issue: PEBBLE-1447