filer: a full remote-deletion tombstone table keeps the lazy fetch off (#12440)

* filer: a full remote-deletion tombstone table keeps the lazy fetch off

At the cap a new delete under a remote mount was dropped with a log line
and isRemoteDeletionPending answered false for it, so the lazy remote fetch
resurrected deleted paths. While the table is full every path may be a
pending delete: isRemoteDeletionPending answers true until the table has
room again. The size and the full state are gauges
(filer_remote_deletion_tombstones, filer_remote_deletion_tombstones_full).

(cherry picked from commit f3be0ab965cde3bfd3053d3ac720fbbe3f68e98b)

* filer: a full tombstone table drains on reads and lifts after the unrecorded deletes age out

While the table is full, a lookup miss evicts expired tombstones and those
the write-back offset has passed, so the table drains without a mutation.
The full state lifts only once the table has room and the newest
unrecorded delete is past the TTL: an unrecorded delete has no event to
confirm it, so only the TTL makes it safe, as for a pending tombstone.

(cherry picked from commit 19394fe5d67798af7b0d755ac18c83c75e20a12b)

* filer: a full tombstone table drains per mount, only when its offset advances

Offset-based release applies to the tombstones under the mount whose
offset was read; another mount's tombstones answer to their own offset. A
read sweeps the table only once that offset advanced since the last drain
for the mount or the unrecorded deletes aged past the TTL, so a listing of
many missing children does not rescan the table per child.

(cherry picked from commit 9cf00e2e54d70e6ba530e6ec1b1ff4b502873847)

* filer: the tombstones-full gauge help names when it clears

---------

Co-authored-by: james <james@medable.com>

weed/filer/filer_remote_tombstone.go

@@ -10,19 +10,26 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/remote_storage"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
 	"google.golang.org/protobuf/proto"
 )
 
 const (
-	// remoteDeletionTombstoneTTL bounds a tombstone when no write-back sync
-	// offset ever confirms the remote delete. A remote object re-created
+	// defaultRemoteDeletionTombstoneTTL bounds a tombstone when no write-back
+	// sync offset ever confirms the remote delete. A remote object re-created
 	// outside the filer at a deleted path stays hidden for this long.
-	remoteDeletionTombstoneTTL = 24 * time.Hour
-	// remoteDeletionTombstoneLimit bounds tracked paths; past it new
-	// tombstones are dropped after an expired sweep still leaves no room.
-	remoteDeletionTombstoneLimit = 1 << 16
+	defaultRemoteDeletionTombstoneTTL = 24 * time.Hour
+	// defaultRemoteDeletionTombstoneLimit bounds tracked paths; at the cap a
+	// new delete is not recorded and the lazy fetch stays off instead.
+	defaultRemoteDeletionTombstoneLimit = 1 << 16
+)
+
+// Variables for tests that fill or age the table.
+var (
+	remoteDeletionTombstoneTTL   = defaultRemoteDeletionTombstoneTTL
+	remoteDeletionTombstoneLimit = defaultRemoteDeletionTombstoneLimit
 )
 
 // remoteDeletionTombstones tracks paths deleted under a remote mount whose
@@ -49,7 +56,92 @@ type remoteDeletionTombstones struct {
 	mu      sync.Mutex
 	files   map[string]remoteDeletionTombstone // file path -> delete event
 	dirs    map[string]remoteDeletionTombstone // deleted directory path -> event; covers its subtree
-	pending map[string]bool                   // tombstone path recorded ahead of its event
+	pending map[string]bool                    // tombstone path recorded ahead of its event
+	// full is set when a delete could not be recorded: from then on an
+	// unrecorded path may be a pending delete, so every lazy fetch stays off.
+	// It lifts once the table has room and the newest unrecorded delete is
+	// past the TTL (skippedThroughTsNs): an unrecorded delete has no event
+	// to confirm it, so only the TTL makes it safe, as for a pending one.
+	full               bool
+	skippedThroughTsNs int64
+	// drainedOffsets records, per mount, the write-back offset the last
+	// drain released through, so a read repeats the sweep only once the
+	// offset has advanced.
+	drainedOffsets map[string]int64
+}
+
+// isFull reports whether an unrecorded delete still keeps the lazy fetch off.
+func (t *remoteDeletionTombstones) isFull() bool {
+	if t == nil {
+		return false
+	}
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.full
+}
+
+// drain, while the table is full, evicts expired tombstones and releases the
+// mount's tombstones its write-back offset has passed, then re-evaluates
+// the full state. It sweeps only when that offset advanced since the last
+// drain for the mount or the unrecorded deletes have aged past the TTL, so
+// a listing of many missing children does not rescan the table per child.
+// haveOffset is false when the mount's persisted offset could not be read.
+func (t *remoteDeletionTombstones) drain(mountDir string, nowNs, offsetTsNs int64, haveOffset bool) {
+	if t == nil {
+		return
+	}
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if t.drainedOffsets == nil {
+		t.drainedOffsets = make(map[string]int64)
+	}
+	advanced := haveOffset && offsetTsNs > t.drainedOffsets[mountDir]
+	aged := t.skippedThroughTsNs != 0 && nowNs-t.skippedThroughTsNs >= int64(remoteDeletionTombstoneTTL)
+	if !advanced && !aged {
+		return
+	}
+	t.evictExpiredLocked(nowNs)
+	if advanced {
+		t.releaseConsumedLocked(mountDir, offsetTsNs)
+		t.drainedOffsets[mountDir] = offsetTsNs
+	}
+	t.observeLocked()
+}
+
+// releaseConsumedLocked drops the mount's tombstones whose event the mount's
+// write-back offset has passed; a pending one has no event timestamp yet
+// and stays, and another mount's tombstones answer to their own offset.
+// Caller must hold t.mu.
+func (t *remoteDeletionTombstones) releaseConsumedLocked(mountDir string, offsetTsNs int64) {
+	prefix := strings.TrimSuffix(mountDir, "/") + "/"
+	under := func(p string) bool { return p == mountDir || strings.HasPrefix(p, prefix) }
+	for p, cur := range t.files {
+		if under(p) && !t.pending[p] && cur.streamTsNs <= offsetTsNs {
+			delete(t.files, p)
+		}
+	}
+	for p, cur := range t.dirs {
+		if under(p) && !t.pending[p] && cur.streamTsNs <= offsetTsNs {
+			delete(t.dirs, p)
+		}
+	}
+}
+
+// observeLocked publishes the table size and lifts the full state once a
+// delete fits again and the newest unrecorded delete is past the TTL.
+// Caller must hold t.mu.
+func (t *remoteDeletionTombstones) observeLocked() {
+	size := len(t.files) + len(t.dirs)
+	if size < remoteDeletionTombstoneLimit && (t.skippedThroughTsNs == 0 || time.Now().UnixNano()-t.skippedThroughTsNs >= int64(remoteDeletionTombstoneTTL)) {
+		t.full = false
+		t.skippedThroughTsNs = 0
+	}
+	stats.FilerRemoteDeletionTombstones.Set(float64(size))
+	if t.full {
+		stats.FilerRemoteDeletionTombstonesFull.Set(1)
+	} else {
+		stats.FilerRemoteDeletionTombstonesFull.Set(0)
+	}
 }
 
 func newRemoteDeletionTombstones() *remoteDeletionTombstones {
@@ -116,7 +208,14 @@ func (t *remoteDeletionTombstones) upsert(path string, isDir bool, commitTsNs, s
 	if len(t.files)+len(t.dirs) >= remoteDeletionTombstoneLimit {
 		t.evictExpiredLocked(time.Now().UnixNano())
 		if len(t.files)+len(t.dirs) >= remoteDeletionTombstoneLimit {
-			glog.V(0).Infof("remote deletion tombstones full (%d), skipping %s", remoteDeletionTombstoneLimit, path)
+			if !t.full {
+				glog.V(0).Infof("remote deletion tombstones full (%d), lazy remote fetch off until the table has room and the unrecorded deletes age out; first skipped %s", remoteDeletionTombstoneLimit, path)
+			}
+			t.full = true
+			if commitTsNs > t.skippedThroughTsNs {
+				t.skippedThroughTsNs = commitTsNs
+			}
+			t.observeLocked()
 			return
 		}
 	}
@@ -127,6 +226,7 @@ func (t *remoteDeletionTombstones) upsert(path string, isDir bool, commitTsNs, s
 	if isDir {
 		t.dropCoveredLocked(path, commitTsNs)
 	}
+	t.observeLocked()
 }
 
 // dropCoveredLocked removes descendant tombstones a new directory tombstone
@@ -165,6 +265,7 @@ func (t *remoteDeletionTombstones) drop(path string, tsNs int64) {
 		delete(t.files, path)
 		delete(t.pending, path)
 	}
+	t.observeLocked()
 }
 
 // clear drops a file tombstone when a write at the path is at least as new as
@@ -179,6 +280,7 @@ func (t *remoteDeletionTombstones) clear(path string, tsNs int64) {
 		delete(t.files, path)
 		delete(t.pending, path)
 	}
+	t.observeLocked()
 }
 
 // blockedSince returns the tombstone governing path — its own file tombstone
@@ -230,6 +332,7 @@ func (t *remoteDeletionTombstones) releaseThrough(path string, tsNs int64) {
 		}
 		p = p[:i]
 	}
+	t.observeLocked()
 }
 
 // releaseExpired drops the tombstones governing path that outlived the TTL —
@@ -255,6 +358,7 @@ func (t *remoteDeletionTombstones) releaseExpired(path string, nowNs int64) {
 		}
 		p = p[:i]
 	}
+	t.observeLocked()
 }
 
 func (t *remoteDeletionTombstones) evictExpiredLocked(nowNs int64) {
@@ -270,6 +374,7 @@ func (t *remoteDeletionTombstones) evictExpiredLocked(nowNs int64) {
 			delete(t.pending, p)
 		}
 	}
+	t.observeLocked()
 }
 
 // noteRemoteDeletion records a delete of path under a remote mount so lazy
@@ -300,6 +405,16 @@ func (f *Filer) isRemoteDeletionPending(ctx context.Context, p util.FullPath, mo
 	if f.remoteTombstones == nil {
 		return false
 	}
+	// With deletes unrecorded, any path may be one: no lazy fetch resurrects
+	// it until the table has room again and the unrecorded deletes aged out.
+	// Reads keep the table draining meanwhile.
+	if f.remoteTombstones.isFull() {
+		offset, err := f.readRemoteSyncOffset(ctx, mountDir)
+		f.remoteTombstones.drain(string(mountDir), time.Now().UnixNano(), offset, err == nil)
+		if f.remoteTombstones.isFull() {
+			return true
+		}
+	}
 	streamTsNs, commitTsNs, pending := f.remoteTombstones.blockedSince(string(p))
 	if commitTsNs == 0 {
 		return false

weed/filer/filer_remote_tombstone_test.go

@@ -165,6 +165,107 @@ func TestRemoteDeletionTombstones_RaisedLocalTombstoneIsPendingAgain(t *testing.
 	assert.False(t, pending)
 }
 
+func TestMaybeLazyFetchFromRemote_FullTableKeepsTheFetchOff(t *testing.T) {
+	const storageType = "stub_tomb_full"
+	stub := &countingRemoteClient{
+		stubRemoteClient: stubRemoteClient{
+			statResult: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 11},
+		},
+	}
+	f, _ := newMountedTestFiler(t, storageType, stub, 0)
+	remoteDeletionTombstoneLimit = 2
+	t.Cleanup(func() { remoteDeletionTombstoneLimit = defaultRemoteDeletionTombstoneLimit })
+
+	now := time.Now().UnixNano()
+	f.noteRemoteDeletion("/buckets/mybucket/dir/a.txt", false, now)
+	f.noteRemoteDeletion("/buckets/mybucket/dir/b.txt", false, now)
+	assert.False(t, f.remoteTombstones.isFull())
+
+	// the third delete does not fit: from here every path may be a pending
+	// delete, so no lookup miss consults the remote
+	f.noteRemoteDeletion("/buckets/mybucket/dir/c.txt", false, now)
+	assert.True(t, f.remoteTombstones.isFull())
+	entry, err := f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/other.txt")
+	require.NoError(t, err)
+	assert.Nil(t, entry)
+	assert.Equal(t, 0, stub.statCalls)
+
+	// room again once a recorded delete lifts, but the unrecorded delete of
+	// c.txt has no event to confirm it: the fetch stays off until the TTL
+	f.remoteTombstones.drop("/buckets/mybucket/dir/a.txt", now)
+	assert.True(t, f.remoteTombstones.isFull())
+	entry, err = f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/other.txt")
+	require.NoError(t, err)
+	assert.Nil(t, entry)
+	assert.Equal(t, 0, stub.statCalls)
+
+	remoteDeletionTombstoneTTL = 0
+	t.Cleanup(func() { remoteDeletionTombstoneTTL = defaultRemoteDeletionTombstoneTTL })
+	entry, err = f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/other.txt")
+	require.NoError(t, err)
+	require.NotNil(t, entry)
+	assert.NotZero(t, stub.statCalls)
+	assert.False(t, f.remoteTombstones.isFull())
+}
+
+func TestMaybeLazyFetchFromRemote_FullTableDrainsOnReads(t *testing.T) {
+	const storageType = "stub_tomb_drain"
+	stub := &countingRemoteClient{
+		stubRemoteClient: stubRemoteClient{
+			statResult: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 11},
+		},
+	}
+	f, store := newMountedTestFiler(t, storageType, stub, 0)
+	remoteDeletionTombstoneLimit = 2
+	t.Cleanup(func() { remoteDeletionTombstoneLimit = defaultRemoteDeletionTombstoneLimit })
+	now := time.Now().UnixNano()
+	tsA, tsB, tsC := now-3*int64(time.Second), now-2*int64(time.Second), now-int64(time.Second)
+
+	// a confirmed delete on another mount, then two on this one fill the
+	// table; a third on this mount is unrecorded and closes the fetch
+	remoteDeletionTombstoneLimit = 3
+	f.remoteTombstones.addFromEvent("/buckets/other/dir/z.txt", false, tsA, tsA)
+	f.remoteTombstones.addFromEvent("/buckets/mybucket/dir/a.txt", false, tsA, tsA)
+	f.remoteTombstones.addFromEvent("/buckets/mybucket/dir/b.txt", false, tsB, tsB)
+	f.remoteTombstones.addFromEvent("/buckets/mybucket/dir/c.txt", false, tsC, tsC)
+	assert.True(t, f.remoteTombstones.isFull())
+	assert.Len(t, f.remoteTombstones.files, 3)
+
+	// the write-back offset behind the recorded deletes: a read drains
+	// nothing and the fetch stays off
+	putRemoteSyncOffset(t, store, "/buckets/mybucket", tsA-1)
+	entry, err := f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/other.txt")
+	require.NoError(t, err)
+	assert.Nil(t, entry)
+	assert.Equal(t, 0, stub.statCalls)
+	assert.Len(t, f.remoteTombstones.files, 3)
+
+	// the offset past both recorded deletes of this mount: the read drains
+	// them without any mutation; the other mount's tombstone answers to its
+	// own offset and stays; the unrecorded delete is still inside the TTL,
+	// so the fetch stays off
+	putRemoteSyncOffset(t, store, "/buckets/mybucket", tsB)
+	entry, err = f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/other.txt")
+	require.NoError(t, err)
+	assert.Nil(t, entry)
+	assert.Equal(t, 0, stub.statCalls)
+	assert.Len(t, f.remoteTombstones.files, 1)
+	_, otherKept := f.remoteTombstones.files["/buckets/other/dir/z.txt"]
+	assert.True(t, otherKept, "another mount's tombstone must not be released by this mount's offset")
+	assert.True(t, f.remoteTombstones.isFull())
+
+	// once the unrecorded delete has aged past the TTL the next read lifts
+	// the gate
+	f.remoteTombstones.mu.Lock()
+	f.remoteTombstones.skippedThroughTsNs = now - int64(remoteDeletionTombstoneTTL) - 1
+	f.remoteTombstones.mu.Unlock()
+	entry, err = f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/other.txt")
+	require.NoError(t, err)
+	require.NotNil(t, entry)
+	assert.NotZero(t, stub.statCalls)
+	assert.False(t, f.remoteTombstones.isFull())
+}
+
 func TestMaybeLazyFetchFromRemote_SkipsTombstonedPath(t *testing.T) {
 	const storageType = "stub_tomb_fetch"
 	stub := &countingRemoteClient{

weed/stats/metrics.go

@@ -300,6 +300,22 @@ var (
 			Help:      "Counter of filer store requests.",
 		}, []string{"store", "type"})
 
+	FilerRemoteDeletionTombstones = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Namespace: Namespace,
+			Subsystem: subsystemFiler,
+			Name:      "remote_deletion_tombstones",
+			Help:      "Paths deleted under a remote mount whose remote delete the write-back daemon has not consumed yet.",
+		})
+
+	FilerRemoteDeletionTombstonesFull = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Namespace: Namespace,
+			Subsystem: subsystemFiler,
+			Name:      "remote_deletion_tombstones_full",
+			Help:      "1 from a delete the full table could not record until the table has room and that delete is past the TTL; lazy remote fetches are off meanwhile. 0 otherwise.",
+		})
+
 	FilerStoreHistogram = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
 			Namespace: Namespace,
@@ -1037,6 +1053,8 @@ func init() {
 	Gather.MustRegister(FilerInFlightUploadCountGauge)
 	Gather.MustRegister(FilerStoreCounter)
 	Gather.MustRegister(FilerStoreHistogram)
+	Gather.MustRegister(FilerRemoteDeletionTombstones)
+	Gather.MustRegister(FilerRemoteDeletionTombstonesFull)
 	Gather.MustRegister(FilerSyncOffsetGauge)
 	Gather.MustRegister(FilerSyncEventsReceivedCounter)
 	Gather.MustRegister(FilerSyncEventsProcessedCounter)