syncWatchers scans unbounded revisions but only delivers 1000, stalling writes

[watsonnsw]

Bug report criteria

• This bug report is not security related, security issues should be disclosed privately via security@etcd.io.
• This is not a support request or question, support requests or questions should be raised in the etcd discussion forums.
• You have read the etcd bug reporting guidelines.
• Existing open issues along with etcd frequently asked questions have been checked and this is not a duplicate.

What happened?

A kubernetes API server's watch fell far behind the tip, becoming unsynced with etcd (over 500,000 revisions behind). Each time the syncWatchers loop fired, a scan of all the missing revisions occurred, but only 1000 revisions were delivered to the watcher per loop due to the hardcoded watchBatchMaxRevs limit. Because the syncWatchers scan holds watchableStore.mu for the full duration and watchableStoreTxnWrite.End() needs the same lock to publish events, this stalls the apply loop for the duration of the recovery. Scan duration reached around 1.3 seconds, with 100 ms between each syncWatchers loop, so writes were blocked for >90% of the time.

What did you expect to happen?

A bounded MVCC scan of only revisions that will actually be sent to the unsynced watcher, with the apply loop not stalled during recovery.

How can we reproduce it (as minimally and precisely as possible)?

Any watcher that is over a thousand revisions behind reproduces this: each pass scans the whole remaining [minRev, curRev] window but delivers at most watchBatchMaxRevs, so the backlog is re-read roughly [number of revisions behind] / 1000 times.

Unit test to repro:

Drop this in server/storage/mvcc/ and run go test ./storage/mvcc/ -run TestSyncWatchersRescansBacklogEveryPass -v. It drives syncWatchers() in a manual loop, so there is no goroutine timing or wall-clock dependence.

package mvcc

import (
	"fmt"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"go.uber.org/zap/zaptest"

	"go.etcd.io/etcd/server/v3/lease"
	betesting "go.etcd.io/etcd/server/v3/storage/backend/testing"
)

func TestSyncWatchersRescansBacklogEveryPass(t *testing.T) {
	b, _ := betesting.NewDefaultTmpBackend(t)
	s := newWatchableStore(zaptest.NewLogger(t), b, &lease.FakeLessor{}, StoreConfig{})
	defer cleanup(s, b)

	// One watcher, backlogRevs behind, on a keyspace of only 50 distinct keys (revision churn,
	// not a large dataset).
	const backlogRevs = 20000
	const keyCount = 50
	putChurn(s, keyCount, backlogRevs)

	w := s.NewWatchStream()
	defer w.Close()
	id, err := w.Watch(t.Context(), 0, []byte("k0"), []byte("k9999"), 1)
	require.NoError(t, err)
	watcher := w.(*watchStream).watchers[id]

	var passes int
	var totalScanned int64
	var totalHold, maxHold time.Duration
	for s.unsynced.size() > 0 {
		// Revisions this pass must read: the scan covers [minRev, curRev] regardless of how
		// many will be delivered.
		scanned := s.Rev() - watcher.minRev + 1

		start := time.Now()
		s.syncWatchers()
		hold := time.Since(start)

		drain(w.(*watchStream))
		passes++
		totalScanned += scanned
		totalHold += hold
		if hold > maxHold {
			maxHold = hold
		}
		if passes <= 3 || s.unsynced.size() == 0 {
			t.Logf("pass %3d: scanned %7d revisions, delivered <=%d, lock held %v",
				passes, scanned, watchBatchMaxRevs, hold.Round(time.Microsecond))
		}
	}

	t.Logf("recovered a watcher %d revisions behind in %d passes", backlogRevs, passes)
	t.Logf("total revisions read: %d to deliver %d (%.0fx amplification)",
		totalScanned, backlogRevs, float64(totalScanned)/float64(backlogRevs))
	t.Logf("watchableStore.mu held for %v total, %v max in a single pass",
		totalHold.Round(time.Millisecond), maxHold.Round(time.Microsecond))

	require.Greater(t, totalScanned, int64(backlogRevs)*5,
		"expected large read amplification from rescanning the backlog every pass")
}

func putChurn(s *watchableStore, keyCount, n int) {
	for i := 0; i < n; i++ {
		s.Put([]byte(fmt.Sprintf("k%d", i%keyCount)), []byte("v"), lease.NoLease)
	}
}

func drain(ws *watchStream) {
	for {
		select {
		case <-ws.ch:
		default:
			return
		}
	}
}

Output on unmodified main (ee043b3), recovering a single watcher 20,000 revisions behind with
no concurrent writes:

pass   1: scanned   20001 revisions, delivered <=1000, lock held 17.568ms
pass   2: scanned   19000 revisions, delivered <=1000, lock held 15.425ms
pass   3: scanned   18000 revisions, delivered <=1000, lock held 15.055ms
...
pass  20: scanned    1000 revisions, delivered <=1000, lock held 587µs
recovered a watcher 20000 revisions behind in 20 passes
total revisions read: 210001 to deliver 20000
watchableStore.mu held for 165ms total, 17.568ms max in a single pass

Anything else we need to know?

I found two previous issues dealing with similar problems: #16839 #18109

It looks like a previously-pursued solution was to cache the scanned revisions, but it looks like it was reverted here: 562f4af (in cases where the number of revisions is very large, this could still cause memory issues).

I am working on a potential solution that instead scans 1000 revisions out from each unsynced watcher. Based on preliminary testing, it looks like bounding the scan like this resolves the latency issue without introducing memory pressure.

Etcd version (please run commands below)

3.5.21 (still present in 3.6.14 and 3.7.1)

Etcd configuration (command line flags or environment variables)

No response

Etcd debug information (please run commands below, feel free to obfuscate the IP address or FQDN in the output)

No response

Relevant log output

[mehrdadbn9]

Hi, I've read through this and I think I can take it on. If it's still open and nobody has picked it up, could you assign it to me? If it's already spoken for or you'd rather it wait for triage, no problem at all - just leave it and I'll find another one.

[watsonnsw]

CC: @hakuna-matatah