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

[RaduBerinde]

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

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

// 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()
}

// 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

// 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

[RaduBerinde]

A possibility for a fix would be to treat the current visibleSeqNum as an implicit snapshot in runDefaultTableCompaction.

[sumeerbhola]

holds for regular writes — keys go through the memtable and visibleSeqNum is bumped before the memtable is flushed into an sstable.

I suspect this is not true either.

This may not be causing a problem in CRDB because we either use ingests during schema change when we are adding completely new userkeys (with at least a different MVCC timestamp), or we are wiping out existing replica state using an incoming range snapshot. For the latter we will need to audit what cases we use a Pebble snapshot from before applying the ingestAndExcise to after (one case may be sending an outgoing range snapshot).

[sumeerbhola]

I gave claude the following hazard, without ingest:

1. Batch commit of a#19. Slow and does not ratchet visibleSeqNum to a#20
2. Flush flushes a#19
3. Compaction C3, starts. It takes a sample of the snapshots, and there is no snapshot at this point in time. It's input includes a#19 which will overwrite a#10.
4. Snapshot takes visibleSeqNum 19. Can see a#10
5. Compaction finishes and has deleted a#10
6. Snapshot can no longer see a#10.

It thinks it is valid:

Walking through it carefully

For your scenario to play out:

• Commit(B) for a#19 runs through commitPipeline.Commit at commit.go:299. The sequence is prepare → env.apply → publish (lines 318/328/337). Between env.apply returning and p.publish(b) being called, the goroutine is just executing two consecutive Go statements — and can be descheduled there (GC pause, OS preemption, anything).
• env.apply calls commitApply which calls mem.apply and then mem.writerUnref() (db.go:866-889). The writerUnref is what eventually allows the memtable to be flushed (readyForFlush is just writerRefs.Load() == 0 at mem_table.go:191). Publish is not on this path.
• So once B has returned from env.apply, B has already done its writerUnref. If the memtable later gets rotated (rotateMemtable at db.go:2916), the queue's writerRef is dropped, and if B was the last in-flight writer the memtable becomes readyForFlush.
• The flush proceeds, produces an sstable containing a#19 (with seqnum 19 — flushes preserve original seqnums).
• A compaction picks up that L0 sstable along with an older L6 sstable containing a#10. C3 captures snapshots := d.mu.snapshots.toSlice() at compaction.go:2720 — empty. compact.Snapshots.IndexAndSeqNum returns SeqNumMax for any key (internal/compact/snapshots.go:59-65). All keys fall in snapshotIdx == 0, and at the bottommost data layer (isBottommostSnapshotStripe returns true at iterator.go:1503-1508) the older entry in the stripe is elided. a#10 is dropped.
• NewSnapshot() (db.go:1832-1849) reads visibleSeqNum.Load() = 19, registers S in d.mu.snapshots. Up to this instant the on-disk Version still has a#10 because C3 hasn't installed yet. loadReadState() from an iterator created now returns the pre-install Version → the iterator can read a#10.
• C3 finishes its I/O, runs UpdateVersionLocked — compact1 at compaction.go:2404-2427 does not consult d.mu.snapshots again. New Version installed. The old Version is unreferenced (modulo any iterator still holding it).
• A new iterator from the same S: loadReadState() returns the new Version, the seqnum-19 filter hides a#19, a#10 is gone. Returns nothing.

Same snapshot, same key, two iterators, two answers. The mechanism is identical to the ingest variant; ingest was just one way to land a key in the LSM with seqnum ≥ visibleSeqNum. A flush of a memtable holding an applied-but-not-yet-published batch does it equally well.

[sumeerbhola]

minor point: we'll want to move the call to maybeScheduleCompaction() to after d.commit.AllocateSeqNum(seqNumCount, prepare, apply), i.e., hoist it out of apply, to reduce the probability that an ingest starts a wasteful compaction with a visibleSeqNum that doesn't include the sstables that were just ingested.